from collections.abc import AsyncIterator

from contextlib import asynccontextmanager

from starlette.applications import Starlette

from starlette.middleware import Middleware

from starlette.middleware.cors import CORSMiddleware

from starlette.routing import Mount

from mcp.server import MCPServer

from mcp.server.transport_security import TransportSecuritySettings

mcp = MCPServer("Notes")

@mcp.tool()

def add_note(text: str) -> str:

"""Save a note."""

return f"Saved: {text}"

@asynccontextmanager

async def lifespan(app: Starlette) -> AsyncIterator[None]:

async with mcp.session_manager.run():

yield

security = TransportSecuritySettings(

allowed_hosts=["mcp.example.com", "mcp.example.com:*"],

allowed_origins=["https://app.example.com"],

)

app = Starlette(

routes=[Mount("/", app=mcp.streamable_http_app(transport_security=security))],

middleware=[

Middleware(

CORSMiddleware,

allow_origins=["https://app.example.com"],

allow_methods=["GET", "POST", "DELETE"],

allow_headers=[

"Authorization",

"Content-Type",

"Last-Event-ID",

"Mcp-Method",

"Mcp-Name",

"Mcp-Protocol-Version",

"Mcp-Session-Id",

],

expose_headers=["Mcp-Session-Id"],

)

],

lifespan=lifespan,

)