modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(examples): migrate all HTTP examples to streamable_http_app() (#2291)

Browse source
This commit is contained in:
Max Isbey 2026-03-16 11:37:01 +00:00 • committed by GitHub
commit abfb482246
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 43 additions and 271 deletions
Download patch file Download diff file Expand all files Collapse all files

29
examples/servers/simple-prompt/mcp_simple_prompt/server.py
View file

@ -2,7 +2,6 @@ import anyio
import click
from mcp import types
from mcp.server import Server, ServerRequestContext
from starlette.requests import Request
def create_messages(context: str | None = None, topic: str | None = None) -> list[types.PromptMessage]:
@ -69,10 +68,10 @@ async def handle_get_prompt(ctx: ServerRequestContext, params: types.GetPromptRe
@click.command()
@click.option("--port", default=8000, help="Port to listen on for SSE")
@click.option("--port", default=8000, help="Port to listen on for HTTP")
@click.option(
"--transport",
type=click.Choice(["stdio", "sse"]),
type=click.Choice(["stdio", "streamable-http"]),
default="stdio",
help="Transport type",
)
@ -83,30 +82,10 @@ def main(port: int, transport: str) -> int:
on_get_prompt=handle_get_prompt,
)
if transport == "sse":
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.responses import Response
from starlette.routing import Mount, Route
sse = SseServerTransport("/messages/")
async def handle_sse(request: Request):
async with sse.connect_sse(request.scope, request.receive, request._send) as streams: # type: ignore[reportPrivateUsage]
await app.run(streams[0], streams[1], app.create_initialization_options())
return Response()
starlette_app = Starlette(
debug=True,
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/messages/", app=sse.handle_post_message),
],
)
if transport == "streamable-http":
import uvicorn
uvicorn.run(starlette_app, host="127.0.0.1", port=port)
uvicorn.run(app.streamable_http_app(), host="127.0.0.1", port=port)
else:
from mcp.server.stdio import stdio_server