#3059·sanic

Sanic 是否支持在 ASGI 模式下安装子应用程序?

作者: Pateo-alandu创建于 2025年4月28日更新于 2026年2月20日
标签feature request

#!/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