"""FastMCP quickstart example.Run from the repository root: uv run examples/snippets/servers/fastmcp_quickstart.py"""frommcp.server.fastmcpimportFastMCP# Create an MCP servermcp=FastMCP("Demo", json_response=True) # Add an addition tool@mcp.tool()defadd(a: int, b: int) ->int: """Add two numbers"""returna+b# Add a dynamic greeting resource@mcp.resource("greeting://{name}")defget_greeting(name: str) ->str: """Get a personalized greeting"""returnf"Hello, {name}!"# Add a prompt@mcp.prompt()defgreet_user(name: str, style: str="friendly") ->str: """Generate a greeting prompt"""styles={"friendly": "Please write a warm, friendly greeting", "formal": "Please write a formal, professional greeting", "casual": "Please write a casual, relaxed greeting", } returnf"{styles.get(style, styles['friendly'])} for someone named {name}."# Run with streamable HTTP transportif__name__=="__main__": mcp.run(transport="streamable-http")

"""Example showing lifespan support for startup/shutdown with strong typing."""fromcollections.abcimportAsyncIteratorfromcontextlibimportasynccontextmanagerfromdataclassesimportdataclassfrommcp.server.fastmcpimportContext, FastMCPfrommcp.server.sessionimportServerSession# Mock database class for exampleclassDatabase: """Mock database class for example."""@classmethodasyncdefconnect(cls) ->"Database": """Connect to database."""returncls() asyncdefdisconnect(self) ->None: """Disconnect from database."""passdefquery(self) ->str: """Execute a query."""return"Query result"@dataclassclassAppContext: """Application context with typed dependencies."""db: Database@asynccontextmanagerasyncdefapp_lifespan(server: FastMCP) ->AsyncIterator[AppContext]: """Manage application lifecycle with type-safe context."""# Initialize on startupdb=awaitDatabase.connect() try: yieldAppContext(db=db) finally: # Cleanup on shutdownawaitdb.disconnect() # Pass lifespan to servermcp=FastMCP("My App", lifespan=app_lifespan) # Access type-safe lifespan context in tools@mcp.tool()defquery_db(ctx: Context[ServerSession, AppContext]) ->str: """Tool that uses initialized resources."""db=ctx.request_context.lifespan_context.dbreturndb.query()

frommcp.server.fastmcpimportFastMCPmcp=FastMCP(name="Resource Example") @mcp.resource("file://documents/{name}")defread_document(name: str) ->str: """Read a document by name."""# This would normally read from diskreturnf"Content of {name}"@mcp.resource("config://settings")defget_settings() ->str: """Get application settings."""return"""{ "theme": "dark", "language": "en", "debug": false}"""

frommcp.server.fastmcpimportFastMCPmcp=FastMCP(name="Tool Example") @mcp.tool()defsum(a: int, b: int) ->int: """Add two numbers together."""returna+b@mcp.tool()defget_weather(city: str, unit: str="celsius") ->str: """Get weather for a city."""# This would normally call a weather APIreturnf"Weather in {city}: 22degrees{unit[0].upper()}"

frommcp.server.fastmcpimportContext, FastMCPfrommcp.server.sessionimportServerSessionmcp=FastMCP(name="Progress Example") @mcp.tool()asyncdeflong_running_task(task_name: str, ctx: Context[ServerSession, None], steps: int=5) ->str: """Execute a task with progress updates."""awaitctx.info(f"Starting: {task_name}") foriinrange(steps): progress= (i+1) /stepsawaitctx.report_progress( progress=progress, total=1.0, message=f"Step {i+1}/{steps}", ) awaitctx.debug(f"Completed step {i+1}") returnf"Task '{task_name}' completed"

"""Example showing direct CallToolResult return for advanced control."""fromtypingimportAnnotatedfrompydanticimportBaseModelfrommcp.server.fastmcpimportFastMCPfrommcp.typesimportCallToolResult, TextContentmcp=FastMCP("CallToolResult Example") classValidationModel(BaseModel): """Model for validating structured output."""status: strdata: dict[str, int] @mcp.tool()defadvanced_tool() ->CallToolResult: """Return CallToolResult directly for full control including _meta field."""returnCallToolResult( content=[TextContent(type="text", text="Response visible to the model")], _meta={"hidden": "data for client applications only"}, ) @mcp.tool()defvalidated_tool() ->Annotated[CallToolResult, ValidationModel]: """Return CallToolResult with structured output validation."""returnCallToolResult( content=[TextContent(type="text", text="Validated response")], structuredContent={"status": "success", "data":{"result": 42}}, _meta={"internal": "metadata"}, ) @mcp.tool()defempty_result_tool() ->CallToolResult: """For empty results, return CallToolResult with empty content."""returnCallToolResult(content=[])

"""Example showing structured output with tools."""fromtypingimportTypedDictfrompydanticimportBaseModel, Fieldfrommcp.server.fastmcpimportFastMCPmcp=FastMCP("Structured Output Example") # Using Pydantic models for rich structured dataclassWeatherData(BaseModel): """Weather information structure."""temperature: float=Field(description="Temperature in Celsius") humidity: float=Field(description="Humidity percentage") condition: strwind_speed: float@mcp.tool()defget_weather(city: str) ->WeatherData: """Get weather for a city - returns structured data."""# Simulated weather datareturnWeatherData( temperature=22.5, humidity=45.0, condition="sunny", wind_speed=5.2, ) # Using TypedDict for simpler structuresclassLocationInfo(TypedDict): latitude: floatlongitude: floatname: str@mcp.tool()defget_location(address: str) ->LocationInfo: """Get location coordinates"""returnLocationInfo(latitude=51.5074, longitude=-0.1278, name="London, UK") # Using dict[str, Any] for flexible schemas@mcp.tool()defget_statistics(data_type: str) ->dict[str, float]: """Get various statistics"""return{"mean": 42.5, "median": 40.0, "std_dev": 5.2} # Ordinary classes with type hints work for structured outputclassUserProfile: name: strage: intemail: str|None=Nonedef__init__(self, name: str, age: int, email: str|None=None): self.name=nameself.age=ageself.email=email@mcp.tool()defget_user(user_id: str) ->UserProfile: """Get user profile - returns structured data"""returnUserProfile(name="Alice", age=30, email="[email protected]") # Classes WITHOUT type hints cannot be used for structured outputclassUntypedConfig: def__init__(self, setting1, setting2): # type: ignore[reportMissingParameterType]self.setting1=setting1self.setting2=setting2@mcp.tool()defget_config() ->UntypedConfig: """This returns unstructured output - no schema generated"""returnUntypedConfig("value1", "value2") # Lists and other types are wrapped automatically@mcp.tool()deflist_cities() ->list[str]: """Get a list of cities"""return ["London", "Paris", "Tokyo"] # Returns:{"result": ["London", "Paris", "Tokyo"]}@mcp.tool()defget_temperature(city: str) ->float: """Get temperature as a simple float"""return22.5# Returns:{"result": 22.5}

frommcp.server.fastmcpimportFastMCPfrommcp.server.fastmcp.promptsimportbasemcp=FastMCP(name="Prompt Example") @mcp.prompt(title="Code Review")defreview_code(code: str) ->str: returnf"Please review this code:\n\n{code}"@mcp.prompt(title="Debug Assistant")defdebug_error(error: str) ->list[base.Message]: return [ base.UserMessage("I'm seeing this error:"), base.UserMessage(error), base.AssistantMessage("I'll help debug that. What have you tried so far?"), ]

frommcp.server.fastmcpimportFastMCP, Icon# Create an icon from a file path or URLicon=Icon( src="icon.png", mimeType="image/png", sizes="64x64" ) # Add icons to servermcp=FastMCP( "My Server", website_url="https://example.com", icons=[icon] ) # Add icons to tools, resources, and prompts@mcp.tool(icons=[icon])defmy_tool(): """Tool with an icon."""return"result"@mcp.resource("demo://resource", icons=[icon])defmy_resource(): """Resource with an icon."""return"content"

"""Example showing image handling with FastMCP."""fromPILimportImageasPILImagefrommcp.server.fastmcpimportFastMCP, Imagemcp=FastMCP("Image Example") @mcp.tool()defcreate_thumbnail(image_path: str) ->Image: """Create a thumbnail from an image"""img=PILImage.open(image_path) img.thumbnail((100, 100)) returnImage(data=img.tobytes(), format="png")

frommcp.server.fastmcpimportContext, FastMCPmcp=FastMCP(name="Context Example") @mcp.tool()asyncdefmy_tool(x: int, ctx: Context) ->str: """Tool that uses context capabilities."""# The context parameter can have any name as long as it's type-annotatedreturnawaitprocess_with_context(x, ctx)

frommcp.server.fastmcpimportContext, FastMCPfrommcp.server.sessionimportServerSessionmcp=FastMCP(name="Progress Example") @mcp.tool()asyncdeflong_running_task(task_name: str, ctx: Context[ServerSession, None], steps: int=5) ->str: """Execute a task with progress updates."""awaitctx.info(f"Starting: {task_name}") foriinrange(steps): progress= (i+1) /stepsawaitctx.report_progress( progress=progress, total=1.0, message=f"Step {i+1}/{steps}", ) awaitctx.debug(f"Completed step {i+1}") returnf"Task '{task_name}' completed"

"""cd to the `examples/snippets` directory and run: uv run completion-client"""importasyncioimportosfrommcpimportClientSession, StdioServerParametersfrommcp.client.stdioimportstdio_clientfrommcp.typesimportPromptReference, ResourceTemplateReference# Create server parameters for stdio connectionserver_params=StdioServerParameters( command="uv", # Using uv to run the serverargs=["run", "server", "completion", "stdio"], # Server with completion supportenv={"UV_INDEX": os.environ.get("UV_INDEX", "")}, ) asyncdefrun(): """Run the completion client example."""asyncwithstdio_client(server_params) as (read, write): asyncwithClientSession(read, write) assession: # Initialize the connectionawaitsession.initialize() # List available resource templatestemplates=awaitsession.list_resource_templates() print("Available resource templates:") fortemplateintemplates.resourceTemplates: print(f" - {template.uriTemplate}") # List available promptsprompts=awaitsession.list_prompts() print("\nAvailable prompts:") forpromptinprompts.prompts: print(f" - {prompt.name}") # Complete resource template argumentsiftemplates.resourceTemplates: template=templates.resourceTemplates[0] print(f"\nCompleting arguments for resource template: {template.uriTemplate}") # Complete without contextresult=awaitsession.complete( ref=ResourceTemplateReference(type="ref/resource", uri=template.uriTemplate), argument={"name": "owner", "value": "model"}, ) print(f"Completions for 'owner' starting with 'model': {result.completion.values}") # Complete with context - repo suggestions based on ownerresult=awaitsession.complete( ref=ResourceTemplateReference(type="ref/resource", uri=template.uriTemplate), argument={"name": "repo", "value": ""}, context_arguments={"owner": "modelcontextprotocol"}, ) print(f"Completions for 'repo' with owner='modelcontextprotocol': {result.completion.values}") # Complete prompt argumentsifprompts.prompts: prompt_name=prompts.prompts[0].nameprint(f"\nCompleting arguments for prompt: {prompt_name}") result=awaitsession.complete( ref=PromptReference(type="ref/prompt", name=prompt_name), argument={"name": "style", "value": ""}, ) print(f"Completions for 'style' argument: {result.completion.values}") defmain(): """Entry point for the completion client."""asyncio.run(run()) if__name__=="__main__": main()

"""Elicitation examples demonstrating form and URL mode elicitation.Form mode elicitation collects structured, non-sensitive data through a schema.URL mode elicitation directs users to external URLs for sensitive operationslike OAuth flows, credential collection, or payment processing."""importuuidfrompydanticimportBaseModel, Fieldfrommcp.server.fastmcpimportContext, FastMCPfrommcp.server.sessionimportServerSessionfrommcp.shared.exceptionsimportUrlElicitationRequiredErrorfrommcp.typesimportElicitRequestURLParamsmcp=FastMCP(name="Elicitation Example") classBookingPreferences(BaseModel): """Schema for collecting user preferences."""checkAlternative: bool=Field(description="Would you like to check another date?") alternativeDate: str=Field( default="2024-12-26", description="Alternative date (YYYY-MM-DD)", ) @mcp.tool()asyncdefbook_table(date: str, time: str, party_size: int, ctx: Context[ServerSession, None]) ->str: """Book a table with date availability check. This demonstrates form mode elicitation for collecting non-sensitive user input. """# Check if date is availableifdate=="2024-12-25": # Date unavailable - ask user for alternativeresult=awaitctx.elicit( message=(f"No tables available for {party_size} on {date}. Would you like to try another date?"), schema=BookingPreferences, ) ifresult.action=="accept"andresult.data: ifresult.data.checkAlternative: returnf"[SUCCESS] Booked for {result.data.alternativeDate}"return"[CANCELLED] No booking made"return"[CANCELLED] Booking cancelled"# Date availablereturnf"[SUCCESS] Booked for {date} at {time}"@mcp.tool()asyncdefsecure_payment(amount: float, ctx: Context[ServerSession, None]) ->str: """Process a secure payment requiring URL confirmation. This demonstrates URL mode elicitation using ctx.elicit_url() for operations that require out-of-band user interaction. """elicitation_id=str(uuid.uuid4()) result=awaitctx.elicit_url( message=f"Please confirm payment of ${amount:.2f}", url=f"https://payments.example.com/confirm?amount={amount}&id={elicitation_id}", elicitation_id=elicitation_id, ) ifresult.action=="accept": # In a real app, the payment confirmation would happen out-of-band# and you'd verify the payment status from your backendreturnf"Payment of ${amount:.2f} initiated - check your browser to complete"elifresult.action=="decline": return"Payment declined by user"return"Payment cancelled"@mcp.tool()asyncdefconnect_service(service_name: str, ctx: Context[ServerSession, None]) ->str: """Connect to a third-party service requiring OAuth authorization. This demonstrates the "throw error" pattern using UrlElicitationRequiredError. Use this pattern when the tool cannot proceed without user authorization. """elicitation_id=str(uuid.uuid4()) # Raise UrlElicitationRequiredError to signal that the client must complete# a URL elicitation before this request can be processed.# The MCP framework will convert this to a -32042 error response.raiseUrlElicitationRequiredError( [ ElicitRequestURLParams( mode="url", message=f"Authorization required to connect to {service_name}", url=f"https://{service_name}.example.com/oauth/authorize?elicit={elicitation_id}", elicitationId=elicitation_id, ) ] )

frommcp.server.fastmcpimportContext, FastMCPfrommcp.server.sessionimportServerSessionfrommcp.typesimportSamplingMessage, TextContentmcp=FastMCP(name="Sampling Example") @mcp.tool()asyncdefgenerate_poem(topic: str, ctx: Context[ServerSession, None]) ->str: """Generate a poem using LLM sampling."""prompt=f"Write a short poem about {topic}"result=awaitctx.session.create_message( messages=[ SamplingMessage( role="user", content=TextContent(type="text", text=prompt), ) ], max_tokens=100, ) # Since we're not passing tools param, result.content is single contentifresult.content.type=="text": returnresult.content.textreturnstr(result.content)

frommcp.server.fastmcpimportContext, FastMCPfrommcp.server.sessionimportServerSessionmcp=FastMCP(name="Notifications Example") @mcp.tool()asyncdefprocess_data(data: str, ctx: Context[ServerSession, None]) ->str: """Process data with logging."""# Different log levelsawaitctx.debug(f"Debug: Processing '{data}'") awaitctx.info("Info: Starting processing") awaitctx.warning("Warning: This is experimental") awaitctx.error("Error: (This is just a demo)") # Notify about resource changesawaitctx.session.send_resource_list_changed() returnf"Processed: {data}"

"""Run from the repository root: uv run examples/snippets/servers/oauth_server.py"""frompydanticimportAnyHttpUrlfrommcp.server.auth.providerimportAccessToken, TokenVerifierfrommcp.server.auth.settingsimportAuthSettingsfrommcp.server.fastmcpimportFastMCPclassSimpleTokenVerifier(TokenVerifier): """Simple token verifier for demonstration."""asyncdefverify_token(self, token: str) ->AccessToken|None: pass# This is where you would implement actual token validation# Create FastMCP instance as a Resource Servermcp=FastMCP( "Weather Service", json_response=True, # Token verifier for authenticationtoken_verifier=SimpleTokenVerifier(), # Auth settings for RFC 9728 Protected Resource Metadataauth=AuthSettings( issuer_url=AnyHttpUrl("https://auth.example.com"), # Authorization Server URLresource_server_url=AnyHttpUrl("http://localhost:3001"), # This server's URLrequired_scopes=["user"], ), ) @mcp.tool()asyncdefget_weather(city: str="London") ->dict[str, str]: """Get weather data for a city"""return{"city": city, "temperature": "22", "condition": "Partly cloudy", "humidity": "65%", } if__name__=="__main__": mcp.run(transport="streamable-http")

@mcp.tool()defserver_info(ctx: Context) ->dict: """Get information about the current server."""return{"name": ctx.fastmcp.name, "instructions": ctx.fastmcp.instructions, "debug_mode": ctx.fastmcp.settings.debug, "log_level": ctx.fastmcp.settings.log_level, "host": ctx.fastmcp.settings.host, "port": ctx.fastmcp.settings.port, }

@mcp.tool()asyncdefnotify_data_update(resource_uri: str, ctx: Context) ->str: """Update data and notify clients of the change."""# Perform data update logic here# Notify clients that this specific resource changedawaitctx.session.send_resource_updated(AnyUrl(resource_uri)) # If this affects the overall resource list, notify about that tooawaitctx.session.send_resource_list_changed() returnf"Updated {resource_uri} and notified clients"

# Example with typed lifespan context@dataclassclassAppContext: db: Databaseconfig: AppConfig@mcp.tool()defquery_with_config(query: str, ctx: Context) ->str: """Execute a query using shared database and configuration."""# Access typed lifespan contextapp_ctx: AppContext=ctx.request_context.lifespan_context# Use shared resourcesconnection=app_ctx.dbsettings=app_ctx.config# Execute query with configurationresult=connection.execute(query, timeout=settings.query_timeout) returnstr(result)

"""Example showing direct execution of an MCP server.This is the simplest way to run an MCP server directly.cd to the `examples/snippets` directory and run: uv run direct-execution-server or python servers/direct_execution.py"""frommcp.server.fastmcpimportFastMCPmcp=FastMCP("My App") @mcp.tool()defhello(name: str="World") ->str: """Say hello to someone."""returnf"Hello, {name}!"defmain(): """Entry point for the direct execution server."""mcp.run() if__name__=="__main__": main()

"""Run from the repository root: uv run examples/snippets/servers/streamable_config.py"""frommcp.server.fastmcpimportFastMCP# Stateless server with JSON responses (recommended)mcp=FastMCP("StatelessServer", stateless_http=True, json_response=True) # Other configuration options:# Stateless server with SSE streaming responses# mcp = FastMCP("StatelessServer", stateless_http=True)# Stateful server with session persistence# mcp = FastMCP("StatefulServer")# Add a simple tool to demonstrate the server@mcp.tool()defgreet(name: str="World") ->str: """Greet someone by name."""returnf"Hello, {name}!"# Run server with streamable_http transportif__name__=="__main__": mcp.run(transport="streamable-http")

"""Run from the repository root: uvicorn examples.snippets.servers.streamable_starlette_mount:app --reload"""importcontextlibfromstarlette.applicationsimportStarlettefromstarlette.routingimportMountfrommcp.server.fastmcpimportFastMCP# Create the Echo serverecho_mcp=FastMCP(name="EchoServer", stateless_http=True, json_response=True) @echo_mcp.tool()defecho(message: str) ->str: """A simple echo tool"""returnf"Echo: {message}"# Create the Math servermath_mcp=FastMCP(name="MathServer", stateless_http=True, json_response=True) @math_mcp.tool()defadd_two(n: int) ->int: """Tool to add two to the input"""returnn+2# Create a combined lifespan to manage both session managers@contextlib.asynccontextmanagerasyncdeflifespan(app: Starlette): asyncwithcontextlib.AsyncExitStack() asstack: awaitstack.enter_async_context(echo_mcp.session_manager.run()) awaitstack.enter_async_context(math_mcp.session_manager.run()) yield# Create the Starlette app and mount the MCP serversapp=Starlette( routes=[ Mount("/echo", echo_mcp.streamable_http_app()), Mount("/math", math_mcp.streamable_http_app()), ], lifespan=lifespan, ) # Note: Clients connect to http://localhost:8000/echo/mcp and http://localhost:8000/math/mcp# To mount at the root of each path (e.g., /echo instead of /echo/mcp):# echo_mcp.settings.streamable_http_path = "/"# math_mcp.settings.streamable_http_path = "/"

fromstarlette.applicationsimportStarlettefromstarlette.middleware.corsimportCORSMiddleware# Create your Starlette app firststarlette_app=Starlette(routes=[...]) # Then wrap it with CORS middlewarestarlette_app=CORSMiddleware( starlette_app, allow_origins=["*"], # Configure appropriately for productionallow_methods=["GET", "POST", "DELETE"], # MCP streamable HTTP methodsexpose_headers=["Mcp-Session-Id"], )

"""Basic example showing how to mount StreamableHTTP server in Starlette.Run from the repository root: uvicorn examples.snippets.servers.streamable_http_basic_mounting:app --reload"""importcontextlibfromstarlette.applicationsimportStarlettefromstarlette.routingimportMountfrommcp.server.fastmcpimportFastMCP# Create MCP servermcp=FastMCP("My App", json_response=True) @mcp.tool()defhello() ->str: """A simple hello tool"""return"Hello from MCP!"# Create a lifespan context manager to run the session manager@contextlib.asynccontextmanagerasyncdeflifespan(app: Starlette): asyncwithmcp.session_manager.run(): yield# Mount the StreamableHTTP server to the existing ASGI serverapp=Starlette( routes=[ Mount("/", app=mcp.streamable_http_app()), ], lifespan=lifespan, )

"""Example showing how to mount StreamableHTTP server using Host-based routing.Run from the repository root: uvicorn examples.snippets.servers.streamable_http_host_mounting:app --reload"""importcontextlibfromstarlette.applicationsimportStarlettefromstarlette.routingimportHostfrommcp.server.fastmcpimportFastMCP# Create MCP servermcp=FastMCP("MCP Host App", json_response=True) @mcp.tool()defdomain_info() ->str: """Get domain-specific information"""return"This is served from mcp.acme.corp"# Create a lifespan context manager to run the session manager@contextlib.asynccontextmanagerasyncdeflifespan(app: Starlette): asyncwithmcp.session_manager.run(): yield# Mount using Host-based routingapp=Starlette( routes=[ Host("mcp.acme.corp", app=mcp.streamable_http_app()), ], lifespan=lifespan, )

"""Example showing how to mount multiple StreamableHTTP servers with path configuration.Run from the repository root: uvicorn examples.snippets.servers.streamable_http_multiple_servers:app --reload"""importcontextlibfromstarlette.applicationsimportStarlettefromstarlette.routingimportMountfrommcp.server.fastmcpimportFastMCP# Create multiple MCP serversapi_mcp=FastMCP("API Server", json_response=True) chat_mcp=FastMCP("Chat Server", json_response=True) @api_mcp.tool()defapi_status() ->str: """Get API status"""return"API is running"@chat_mcp.tool()defsend_message(message: str) ->str: """Send a chat message"""returnf"Message sent: {message}"# Configure servers to mount at the root of each path# This means endpoints will be at /api and /chat instead of /api/mcp and /chat/mcpapi_mcp.settings.streamable_http_path="/"chat_mcp.settings.streamable_http_path="/"# Create a combined lifespan to manage both session managers@contextlib.asynccontextmanagerasyncdeflifespan(app: Starlette): asyncwithcontextlib.AsyncExitStack() asstack: awaitstack.enter_async_context(api_mcp.session_manager.run()) awaitstack.enter_async_context(chat_mcp.session_manager.run()) yield# Mount the serversapp=Starlette( routes=[ Mount("/api", app=api_mcp.streamable_http_app()), Mount("/chat", app=chat_mcp.streamable_http_app()), ], lifespan=lifespan, )

"""Example showing path configuration during FastMCP initialization.Run from the repository root: uvicorn examples.snippets.servers.streamable_http_path_config:app --reload"""fromstarlette.applicationsimportStarlettefromstarlette.routingimportMountfrommcp.server.fastmcpimportFastMCP# Configure streamable_http_path during initialization# This server will mount at the root of wherever it's mountedmcp_at_root=FastMCP( "My Server", json_response=True, streamable_http_path="/", ) @mcp_at_root.tool()defprocess_data(data: str) ->str: """Process some data"""returnf"Processed: {data}"# Mount at /process - endpoints will be at /process instead of /process/mcpapp=Starlette( routes=[ Mount("/process", app=mcp_at_root.streamable_http_app()), ] )

fromstarlette.applicationsimportStarlettefromstarlette.routingimportMount, Hostfrommcp.server.fastmcpimportFastMCPmcp=FastMCP("My App") # Mount the SSE server to the existing ASGI serverapp=Starlette( routes=[ Mount('/', app=mcp.sse_app()), ] ) # or dynamically mount as hostapp.router.routes.append(Host('mcp.acme.corp', app=mcp.sse_app()))

fromstarlette.applicationsimportStarlettefromstarlette.routingimportMountfrommcp.server.fastmcpimportFastMCP# Create multiple MCP serversgithub_mcp=FastMCP("GitHub API") browser_mcp=FastMCP("Browser") curl_mcp=FastMCP("Curl") search_mcp=FastMCP("Search") # Method 1: Configure mount paths via settings (recommended for persistent configuration)github_mcp.settings.mount_path="/github"browser_mcp.settings.mount_path="/browser"# Method 2: Pass mount path directly to sse_app (preferred for ad-hoc mounting)# This approach doesn't modify the server's settings permanently# Create Starlette app with multiple mounted serversapp=Starlette( routes=[ # Using settings-based configurationMount("/github", app=github_mcp.sse_app()), Mount("/browser", app=browser_mcp.sse_app()), # Using direct mount path parameterMount("/curl", app=curl_mcp.sse_app("/curl")), Mount("/search", app=search_mcp.sse_app("/search")), ] ) # Method 3: For direct execution, you can also pass the mount path to run()if__name__=="__main__": search_mcp.run(transport="sse", mount_path="/search")

"""Run from the repository root: uv run examples/snippets/servers/lowlevel/lifespan.py"""fromcollections.abcimportAsyncIteratorfromcontextlibimportasynccontextmanagerfromtypingimportAnyimportmcp.server.stdioimportmcp.typesastypesfrommcp.server.lowlevelimportNotificationOptions, Serverfrommcp.server.modelsimportInitializationOptions# Mock database class for exampleclassDatabase: """Mock database class for example."""@classmethodasyncdefconnect(cls) ->"Database": """Connect to database."""print("Database connected") returncls() asyncdefdisconnect(self) ->None: """Disconnect from database."""print("Database disconnected") asyncdefquery(self, query_str: str) ->list[dict[str, str]]: """Execute a query."""# Simulate database queryreturn [{"id": "1", "name": "Example", "query": query_str}] @asynccontextmanagerasyncdefserver_lifespan(_server: Server) ->AsyncIterator[dict[str, Any]]: """Manage server startup and shutdown lifecycle."""# Initialize resources on startupdb=awaitDatabase.connect() try: yield{"db": db} finally: # Clean up on shutdownawaitdb.disconnect() # Pass lifespan to serverserver=Server("example-server", lifespan=server_lifespan) @server.list_tools()asyncdefhandle_list_tools() ->list[types.Tool]: """List available tools."""return [ types.Tool( name="query_db", description="Query the database", inputSchema={"type": "object", "properties":{"query":{"type": "string", "description": "SQL query to execute"}}, "required": ["query"], }, ) ] @server.call_tool()asyncdefquery_db(name: str, arguments: dict[str, Any]) ->list[types.TextContent]: """Handle database query tool call."""ifname!="query_db": raiseValueError(f"Unknown tool: {name}") # Access lifespan contextctx=server.request_contextdb=ctx.lifespan_context["db"] # Execute queryresults=awaitdb.query(arguments["query"]) return [types.TextContent(type="text", text=f"Query results: {results}")] asyncdefrun(): """Run the server with lifespan management."""asyncwithmcp.server.stdio.stdio_server() as (read_stream, write_stream): awaitserver.run( read_stream, write_stream, InitializationOptions( server_name="example-server", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), ) if__name__=="__main__": importasyncioasyncio.run(run())

"""Run from the repository root:uv run examples/snippets/servers/lowlevel/basic.py"""importasyncioimportmcp.server.stdioimportmcp.typesastypesfrommcp.server.lowlevelimportNotificationOptions, Serverfrommcp.server.modelsimportInitializationOptions# Create a server instanceserver=Server("example-server") @server.list_prompts()asyncdefhandle_list_prompts() ->list[types.Prompt]: """List available prompts."""return [ types.Prompt( name="example-prompt", description="An example prompt template", arguments=[types.PromptArgument(name="arg1", description="Example argument", required=True)], ) ] @server.get_prompt()asyncdefhandle_get_prompt(name: str, arguments: dict[str, str] |None) ->types.GetPromptResult: """Get a specific prompt by name."""ifname!="example-prompt": raiseValueError(f"Unknown prompt: {name}") arg1_value= (argumentsor{}).get("arg1", "default") returntypes.GetPromptResult( description="Example prompt", messages=[ types.PromptMessage( role="user", content=types.TextContent(type="text", text=f"Example prompt text with argument: {arg1_value}"), ) ], ) asyncdefrun(): """Run the basic low-level server."""asyncwithmcp.server.stdio.stdio_server() as (read_stream, write_stream): awaitserver.run( read_stream, write_stream, InitializationOptions( server_name="example", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), ) if__name__=="__main__": asyncio.run(run())

"""Run from the repository root: uv run examples/snippets/servers/lowlevel/structured_output.py"""importasynciofromtypingimportAnyimportmcp.server.stdioimportmcp.typesastypesfrommcp.server.lowlevelimportNotificationOptions, Serverfrommcp.server.modelsimportInitializationOptionsserver=Server("example-server") @server.list_tools()asyncdeflist_tools() ->list[types.Tool]: """List available tools with structured output schemas."""return [ types.Tool( name="get_weather", description="Get current weather for a city", inputSchema={"type": "object", "properties":{"city":{"type": "string", "description": "City name"}}, "required": ["city"], }, outputSchema={"type": "object", "properties":{"temperature":{"type": "number", "description": "Temperature in Celsius"}, "condition":{"type": "string", "description": "Weather condition"}, "humidity":{"type": "number", "description": "Humidity percentage"}, "city":{"type": "string", "description": "City name"}, }, "required": ["temperature", "condition", "humidity", "city"], }, ) ] @server.call_tool()asyncdefcall_tool(name: str, arguments: dict[str, Any]) ->dict[str, Any]: """Handle tool calls with structured output."""ifname=="get_weather": city=arguments["city"] # Simulated weather data - in production, call a weather APIweather_data={"temperature": 22.5, "condition": "partly cloudy", "humidity": 65, "city": city, # Include the requested city } # low-level server will validate structured output against the tool's# output schema, and additionally serialize it into a TextContent block# for backwards compatibility with pre-2025-06-18 clients.returnweather_dataelse: raiseValueError(f"Unknown tool: {name}") asyncdefrun(): """Run the structured output server."""asyncwithmcp.server.stdio.stdio_server() as (read_stream, write_stream): awaitserver.run( read_stream, write_stream, InitializationOptions( server_name="structured-output-example", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), ) if__name__=="__main__": asyncio.run(run())

"""Run from the repository root: uv run examples/snippets/servers/lowlevel/direct_call_tool_result.py"""importasynciofromtypingimportAnyimportmcp.server.stdioimportmcp.typesastypesfrommcp.server.lowlevelimportNotificationOptions, Serverfrommcp.server.modelsimportInitializationOptionsserver=Server("example-server") @server.list_tools()asyncdeflist_tools() ->list[types.Tool]: """List available tools."""return [ types.Tool( name="advanced_tool", description="Tool with full control including _meta field", inputSchema={"type": "object", "properties":{"message":{"type": "string"}}, "required": ["message"], }, ) ] @server.call_tool()asyncdefhandle_call_tool(name: str, arguments: dict[str, Any]) ->types.CallToolResult: """Handle tool calls by returning CallToolResult directly."""ifname=="advanced_tool": message=str(arguments.get("message", "")) returntypes.CallToolResult( content=[types.TextContent(type="text", text=f"Processed: {message}")], structuredContent={"result": "success", "message": message}, _meta={"hidden": "data for client applications only"}, ) raiseValueError(f"Unknown tool: {name}") asyncdefrun(): """Run the server."""asyncwithmcp.server.stdio.stdio_server() as (read_stream, write_stream): awaitserver.run( read_stream, write_stream, InitializationOptions( server_name="example", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), ) if__name__=="__main__": asyncio.run(run())

"""Example of implementing pagination with MCP server decorators."""frompydanticimportAnyUrlimportmcp.typesastypesfrommcp.server.lowlevelimportServer# Initialize the serverserver=Server("paginated-server") # Sample data to paginateITEMS= [f"Item {i}"foriinrange(1, 101)] # 100 items@server.list_resources()asyncdeflist_resources_paginated(request: types.ListResourcesRequest) ->types.ListResourcesResult: """List resources with pagination support."""page_size=10# Extract cursor from request paramscursor=request.params.cursorifrequest.paramsisnotNoneelseNone# Parse cursor to get offsetstart=0ifcursorisNoneelseint(cursor) end=start+page_size# Get page of resourcespage_items= [ types.Resource(uri=AnyUrl(f"resource://items/{item}"), name=item, description=f"Description for {item}") foriteminITEMS[start:end] ] # Determine next cursornext_cursor=str(end) ifend<len(ITEMS) elseNonereturntypes.ListResourcesResult(resources=page_items, nextCursor=next_cursor)

"""Example of consuming paginated MCP endpoints from a client."""importasynciofrommcp.client.sessionimportClientSessionfrommcp.client.stdioimportStdioServerParameters, stdio_clientfrommcp.typesimportPaginatedRequestParams, Resourceasyncdeflist_all_resources() ->None: """Fetch all resources using pagination."""asyncwithstdio_client(StdioServerParameters(command="uv", args=["run", "mcp-simple-pagination"])) as ( read, write, ): asyncwithClientSession(read, write) assession: awaitsession.initialize() all_resources: list[Resource] = [] cursor=NonewhileTrue: # Fetch a page of resourcesresult=awaitsession.list_resources(params=PaginatedRequestParams(cursor=cursor)) all_resources.extend(result.resources) print(f"Fetched {len(result.resources)} resources") # Check if there are more pagesifresult.nextCursor: cursor=result.nextCursorelse: breakprint(f"Total resources: {len(all_resources)}") if__name__=="__main__": asyncio.run(list_all_resources())

"""cd to the `examples/snippets/clients` directory and run: uv run client"""importasyncioimportosfrompydanticimportAnyUrlfrommcpimportClientSession, StdioServerParameters, typesfrommcp.client.stdioimportstdio_clientfrommcp.shared.contextimportRequestContext# Create server parameters for stdio connectionserver_params=StdioServerParameters( command="uv", # Using uv to run the serverargs=["run", "server", "fastmcp_quickstart", "stdio"], # We're already in snippets direnv={"UV_INDEX": os.environ.get("UV_INDEX", "")}, ) # Optional: create a sampling callbackasyncdefhandle_sampling_message( context: RequestContext[ClientSession, None], params: types.CreateMessageRequestParams ) ->types.CreateMessageResult: print(f"Sampling request: {params.messages}") returntypes.CreateMessageResult( role="assistant", content=types.TextContent( type="text", text="Hello, world! from model", ), model="gpt-3.5-turbo", stopReason="endTurn", ) asyncdefrun(): asyncwithstdio_client(server_params) as (read, write): asyncwithClientSession(read, write, sampling_callback=handle_sampling_message) assession: # Initialize the connectionawaitsession.initialize() # List available promptsprompts=awaitsession.list_prompts() print(f"Available prompts: {[p.nameforpinprompts.prompts]}") # Get a prompt (greet_user prompt from fastmcp_quickstart)ifprompts.prompts: prompt=awaitsession.get_prompt("greet_user", arguments={"name": "Alice", "style": "friendly"}) print(f"Prompt result: {prompt.messages[0].content}") # List available resourcesresources=awaitsession.list_resources() print(f"Available resources: {[r.uriforrinresources.resources]}") # List available toolstools=awaitsession.list_tools() print(f"Available tools: {[t.namefortintools.tools]}") # Read a resource (greeting resource from fastmcp_quickstart)resource_content=awaitsession.read_resource(AnyUrl("greeting://World")) content_block=resource_content.contents[0] ifisinstance(content_block, types.TextContent): print(f"Resource content: {content_block.text}") # Call a tool (add tool from fastmcp_quickstart)result=awaitsession.call_tool("add", arguments={"a": 5, "b": 3}) result_unstructured=result.content[0] ifisinstance(result_unstructured, types.TextContent): print(f"Tool result: {result_unstructured.text}") result_structured=result.structuredContentprint(f"Structured tool result: {result_structured}") defmain(): """Entry point for the client script."""asyncio.run(run()) if__name__=="__main__": main()

"""Run from the repository root: uv run examples/snippets/clients/streamable_basic.py"""importasynciofrommcpimportClientSessionfrommcp.client.streamable_httpimportstreamable_http_clientasyncdefmain(): # Connect to a streamable HTTP serverasyncwithstreamable_http_client("http://localhost:8000/mcp") as ( read_stream, write_stream, _, ): # Create a session using the client streamsasyncwithClientSession(read_stream, write_stream) assession: # Initialize the connectionawaitsession.initialize() # List available toolstools=awaitsession.list_tools() print(f"Available tools: {[tool.namefortoolintools.tools]}") if__name__=="__main__": asyncio.run(main())

"""cd to the `examples/snippets` directory and run: uv run display-utilities-client"""importasyncioimportosfrommcpimportClientSession, StdioServerParametersfrommcp.client.stdioimportstdio_clientfrommcp.shared.metadata_utilsimportget_display_name# Create server parameters for stdio connectionserver_params=StdioServerParameters( command="uv", # Using uv to run the serverargs=["run", "server", "fastmcp_quickstart", "stdio"], env={"UV_INDEX": os.environ.get("UV_INDEX", "")}, ) asyncdefdisplay_tools(session: ClientSession): """Display available tools with human-readable names"""tools_response=awaitsession.list_tools() fortoolintools_response.tools: # get_display_name() returns the title if available, otherwise the namedisplay_name=get_display_name(tool) print(f"Tool: {display_name}") iftool.description: print(f" {tool.description}") asyncdefdisplay_resources(session: ClientSession): """Display available resources with human-readable names"""resources_response=awaitsession.list_resources() forresourceinresources_response.resources: display_name=get_display_name(resource) print(f"Resource: {display_name} ({resource.uri})") templates_response=awaitsession.list_resource_templates() fortemplateintemplates_response.resourceTemplates: display_name=get_display_name(template) print(f"Resource Template: {display_name}") asyncdefrun(): """Run the display utilities example."""asyncwithstdio_client(server_params) as (read, write): asyncwithClientSession(read, write) assession: # Initialize the connectionawaitsession.initialize() print("=== Available Tools ===") awaitdisplay_tools(session) print("\n=== Available Resources ===") awaitdisplay_resources(session) defmain(): """Entry point for the display utilities client."""asyncio.run(run()) if__name__=="__main__": main()

"""Before running, specify running MCP RS server URL.To spin up RS server locally, see examples/servers/simple-auth/README.mdcd to the `examples/snippets` directory and run: uv run oauth-client"""importasynciofromurllib.parseimportparse_qs, urlparseimporthttpxfrompydanticimportAnyUrlfrommcpimportClientSessionfrommcp.client.authimportOAuthClientProvider, TokenStoragefrommcp.client.streamable_httpimportstreamable_http_clientfrommcp.shared.authimportOAuthClientInformationFull, OAuthClientMetadata, OAuthTokenclassInMemoryTokenStorage(TokenStorage): """Demo In-memory token storage implementation."""def__init__(self): self.tokens: OAuthToken|None=Noneself.client_info: OAuthClientInformationFull|None=Noneasyncdefget_tokens(self) ->OAuthToken|None: """Get stored tokens."""returnself.tokensasyncdefset_tokens(self, tokens: OAuthToken) ->None: """Store tokens."""self.tokens=tokensasyncdefget_client_info(self) ->OAuthClientInformationFull|None: """Get stored client information."""returnself.client_infoasyncdefset_client_info(self, client_info: OAuthClientInformationFull) ->None: """Store client information."""self.client_info=client_infoasyncdefhandle_redirect(auth_url: str) ->None: print(f"Visit: {auth_url}") asyncdefhandle_callback() ->tuple[str, str|None]: callback_url=input("Paste callback URL: ") params=parse_qs(urlparse(callback_url).query) returnparams["code"][0], params.get("state", [None])[0] asyncdefmain(): """Run the OAuth client example."""oauth_auth=OAuthClientProvider( server_url="http://localhost:8001", client_metadata=OAuthClientMetadata( client_name="Example MCP Client", redirect_uris=[AnyUrl("http://localhost:3000/callback")], grant_types=["authorization_code", "refresh_token"], response_types=["code"], scope="user", ), storage=InMemoryTokenStorage(), redirect_handler=handle_redirect, callback_handler=handle_callback, ) asyncwithhttpx.AsyncClient(auth=oauth_auth, follow_redirects=True) ascustom_client: asyncwithstreamable_http_client("http://localhost:8001/mcp", http_client=custom_client) as (read, write, _): asyncwithClientSession(read, write) assession: awaitsession.initialize() tools=awaitsession.list_tools() print(f"Available tools: {[tool.namefortoolintools.tools]}") resources=awaitsession.list_resources() print(f"Available resources: {[r.uriforrinresources.resources]}") defrun(): asyncio.run(main()) if__name__=="__main__": run()

"""examples/snippets/clients/parsing_tool_results.py"""importasynciofrommcpimportClientSession, StdioServerParameters, typesfrommcp.client.stdioimportstdio_clientasyncdefparse_tool_results(): """Demonstrates how to parse different types of content in CallToolResult."""server_params=StdioServerParameters( command="python", args=["path/to/mcp_server.py"] ) asyncwithstdio_client(server_params) as (read, write): asyncwithClientSession(read, write) assession: awaitsession.initialize() # Example 1: Parsing text contentresult=awaitsession.call_tool("get_data",{"format": "text"}) forcontentinresult.content: ifisinstance(content, types.TextContent): print(f"Text: {content.text}") # Example 2: Parsing structured content from JSON toolsresult=awaitsession.call_tool("get_user",{"id": "123"}) ifhasattr(result, "structuredContent") andresult.structuredContent: # Access structured data directlyuser_data=result.structuredContentprint(f"User: {user_data.get('name')}, Age: {user_data.get('age')}") # Example 3: Parsing embedded resourcesresult=awaitsession.call_tool("read_config",{}) forcontentinresult.content: ifisinstance(content, types.EmbeddedResource): resource=content.resourceifisinstance(resource, types.TextResourceContents): print(f"Config from {resource.uri}: {resource.text}") elifisinstance(resource, types.BlobResourceContents): print(f"Binary data from {resource.uri}") # Example 4: Parsing image contentresult=awaitsession.call_tool("generate_chart",{"data": [1, 2, 3]}) forcontentinresult.content: ifisinstance(content, types.ImageContent): print(f"Image ({content.mimeType}): {len(content.data)} bytes") # Example 5: Handling errorsresult=awaitsession.call_tool("failing_tool",{}) ifresult.isError: print("Tool execution failed!") forcontentinresult.content: ifisinstance(content, types.TextContent): print(f"Error: {content.text}") asyncdefmain(): awaitparse_tool_results() if__name__=="__main__": asyncio.run(main())