Sanic 是否支持在 ASGI 模式下安装子应用程序?
#!/usr/bin/env python3
-- coding:utf-8 --
import signal import uvicorn from sanic import Sanic from sanic.request import Request from sanic.log import logger from mcp.server.fastmcp import FastMCP from mcp.server.sse import SseServerTransport
app = Sanic(name)
mcp = FastMCP("Sample MCP Server")
sse = SseServerTransport("/messages/")
@mcp.tool() def hello_world(name: str) -> str: """ Given a person's name, output a greeting
input:
name: str -> a person's name
output:
str -> a greeting
"""
return f"hello, {name}!"@app.route("/messages/", methods=["POST"], stream=True) async def mcp_messages_handle(request: Request): asgi_scope = request.app._asgi_app.transport.scope # type: ignore asgi_receive = request.app._asgi_app.transport._receive # type: ignore asgi_send = request.app._asgi_app.transport._send # type: ignore
await sse.handle_post_message(asgi_scope, asgi_receive, asgi_send)@app.route("/sse", methods=["GET"]) async def mcp_sse_handle(request: Request): asgi_scope = request.app._asgi_app.transport.scope # type: ignore asgi_receive = request.app._asgi_app.transport._receive # type: ignore asgi_send = request.app._asgi_app.transport._send # type: ignore
async with sse.connect_sse(asgi_scope, asgi_receive, asgi_send) as (
read_stream,
write_stream,
):
# Run the MCP server with the established streams
await mcp._mcp_server.run(
read_stream,
write_stream,
mcp._mcp_server.create_initialization_options()
)def stopServer(signal, frame): logger.info("Stopping Sanic Server...") app.stop() logger.info("Stoped Sanic Server!")
if name == "main": logger.info("Starting Sanic Server...")
signal.signal(signal.SIGINT, lambda signal, frame: stopServer(signal, frame))
signal.signal(signal.SIGTERM, lambda signal, frame: stopServer(signal, frame))
…
内容来源: sanic-org/sanic