# server.pyfrommcp.server.fastmcpimportFastMCP# Create an MCP servermcp=FastMCP("Demo") # 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 lifespan support for startup/shutdown with strong typingfromcontextlibimportasynccontextmanagerfromcollections.abcimportAsyncIteratorfromdataclassesimportdataclassfromfake_databaseimportDatabase# Replace with your actual DB typefrommcp.server.fastmcpimportContext, FastMCP# Create a named servermcp=FastMCP("My App") # Specify dependencies for deployment and developmentmcp=FastMCP("My App", dependencies=["pandas", "numpy"]) @dataclassclassAppContext: 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) ->str: """Tool that uses initialized resources"""db=ctx.request_context.lifespan_context.dbreturndb.query()

frommcp.server.fastmcpimportFastMCPmcp=FastMCP("My App") @mcp.resource("config://app")defget_config() ->str: """Static configuration data"""return"App configuration here"@mcp.resource("users://{user_id}/profile")defget_user_profile(user_id: str) ->str: """Dynamic user data"""returnf"Profile data for user {user_id}"

importhttpxfrommcp.server.fastmcpimportFastMCPmcp=FastMCP("My App") @mcp.tool()defcalculate_bmi(weight_kg: float, height_m: float) ->float: """Calculate BMI given weight in kg and height in meters"""returnweight_kg/ (height_m**2) @mcp.tool()asyncdeffetch_weather(city: str) ->str: """Fetch current weather for a city"""asyncwithhttpx.AsyncClient() asclient: response=awaitclient.get(f"https://api.weather.com/{city}") returnresponse.text

frommcp.server.fastmcpimportFastMCPfrommcp.server.fastmcp.promptsimportbasemcp=FastMCP("My App") @mcp.prompt()defreview_code(code: str) ->str: returnf"Please review this code:\n\n{code}"@mcp.prompt()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, ImagefromPILimportImageasPILImagemcp=FastMCP("My App") @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.fastmcpimportFastMCP, Contextmcp=FastMCP("My App") @mcp.tool()asyncdeflong_task(files: list[str], ctx: Context) ->str: """Process multiple files with progress tracking"""fori, fileinenumerate(files): ctx.info(f"Processing {file}") awaitctx.report_progress(i, len(files)) data, mime_type=awaitctx.read_resource(f"file://{file}") return"Processing complete"

frommcp.server.fastmcpimportFastMCPmcp=FastMCP("My App") if__name__=="__main__": mcp.run()

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()))

frommcp.server.fastmcpimportFastMCPmcp=FastMCP("Echo") @mcp.resource("echo://{message}")defecho_resource(message: str) ->str: """Echo a message as a resource"""returnf"Resource echo: {message}"@mcp.tool()defecho_tool(message: str) ->str: """Echo a message as a tool"""returnf"Tool echo: {message}"@mcp.prompt()defecho_prompt(message: str) ->str: """Create an echo prompt"""returnf"Please process this message: {message}"

importsqlite3frommcp.server.fastmcpimportFastMCPmcp=FastMCP("SQLite Explorer") @mcp.resource("schema://main")defget_schema() ->str: """Provide the database schema as a resource"""conn=sqlite3.connect("database.db") schema=conn.execute("SELECT sql FROM sqlite_master WHERE type='table'").fetchall() return"\n".join(sql[0] forsqlinschemaifsql[0]) @mcp.tool()defquery_data(sql: str) ->str: """Execute SQL queries safely"""conn=sqlite3.connect("database.db") try: result=conn.execute(sql).fetchall() return"\n".join(str(row) forrowinresult) exceptExceptionase: returnf"Error: {str(e)}"

fromcontextlibimportasynccontextmanagerfromcollections.abcimportAsyncIteratorfromfake_databaseimportDatabase# Replace with your actual DB typefrommcp.serverimportServer@asynccontextmanagerasyncdefserver_lifespan(server: Server) ->AsyncIterator[dict]: """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) # Access lifespan context in handlers@server.call_tool()asyncdefquery_db(name: str, arguments: dict) ->list: ctx=server.request_contextdb=ctx.lifespan_context["db"] returnawaitdb.query(arguments["query"])

importmcp.server.stdioimportmcp.typesastypesfrommcp.server.lowlevelimportNotificationOptions, Serverfrommcp.server.modelsimportInitializationOptions# Create a server instanceserver=Server("example-server") @server.list_prompts()asyncdefhandle_list_prompts() ->list[types.Prompt]: 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: ifname!="example-prompt": raiseValueError(f"Unknown prompt: {name}") returntypes.GetPromptResult( description="Example prompt", messages=[ types.PromptMessage( role="user", content=types.TextContent(type="text", text="Example prompt text"), ) ], ) asyncdefrun(): 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__": importasyncioasyncio.run(run())

frommcpimportClientSession, StdioServerParameters, typesfrommcp.client.stdioimportstdio_client# Create server parameters for stdio connectionserver_params=StdioServerParameters( command="python", # Executableargs=["example_server.py"], # Optional command line argumentsenv=None, # Optional environment variables ) # Optional: create a sampling callbackasyncdefhandle_sampling_message( message: types.CreateMessageRequestParams, ) ->types.CreateMessageResult: 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() # Get a promptprompt=awaitsession.get_prompt( "example-prompt", arguments={"arg1": "value"} ) # List available resourcesresources=awaitsession.list_resources() # List available toolstools=awaitsession.list_tools() # Read a resourcecontent, mime_type=awaitsession.read_resource("file://some/path") # Call a toolresult=awaitsession.call_tool("tool-name", arguments={"arg1": "value"}) if__name__=="__main__": importasyncioasyncio.run(run())