Saltar a contenido

Añadir a una app existente

Traducción automática

Esta página se tradujo automáticamente a partir de la documentación en inglés, y la página en inglés es la versión de referencia. Si algo no se lee bien, Traducciones explica cómo avisarnos.

mcp.run("streamable-http") arranca un servidor web por ti. A veces no es lo que quieres: el servidor MCP es una pieza de una aplicación web más grande, o ya tienes un despliegue ASGI.

Para eso, mcp.streamable_http_app() devuelve una aplicación Starlette.

Una app Starlette es una app ASGI, así que cualquier cosa que aloje ASGI (uvicorn, Hypercorn, otra Starlette, FastAPI) puede alojar el servidor MCP.

La app

server.py
from mcp.server import MCPServer

mcp = MCPServer("Notes")


@mcp.tool()
def add_note(text: str) -> str:
    """Save a note."""
    return f"Saved: {text}"


app = mcp.streamable_http_app()

app es una aplicación ASGI corriente. Pásala a cualquier servidor ASGI:

uvicorn server:app

El endpoint MCP está en /mcp, así que un cliente se conecta a http://127.0.0.1:8000/mcp.

La app ya trae dos cosas:

  • Una ruta, /mcp: el endpoint Streamable HTTP.
  • Un lifespan (ciclo de vida del servidor) que arranca mcp.session_manager, el objeto que se encarga del trabajo en segundo plano de cada sesión activa.

Ejecuta la app por sí sola (uvicorn server:app) y nunca tendrás que pensar en ninguna de las dos.

Tip

streamable_http_app() acepta los mismos argumentos nombrados que mcp.run("streamable-http", ...), menos port: el puerto es cosa de lo que sirva la app. host se sigue aceptando, pero aquí no enlaza nada; Desplegar y escalar explica qué controla realmente. Ejecutar el servidor cubre las opciones en sí.

mcp.sse_app() hace lo mismo para el transporte SSE, ya reemplazado.

Solo localhost, hasta que digas lo contrario

Por defecto, la app responde solo a las solicitudes dirigidas a localhost. streamable_http_app() no puede saber detrás de qué nombre de host se va a servir, así que activa la protección contra DNS rebinding con la lista de permitidos más segura posible; en tu máquina eso es justo lo correcto. Desplegada detrás de un nombre de host real, significa que toda solicitud se rechaza con 421 Misdirected Request hasta que le pases a transport_security= una lista de permitidos con lo que realmente sirves. Nada de lo que construiste llega siquiera a consultarse antes. Esa lista de permitidos, y todo lo demás que hay entre una app que funciona y un nombre de host real, está en Desplegar y escalar.

Montarla

En cuanto el servidor MCP es parte de una aplicación más grande, metes la app dentro de un Mount. Y en cuanto haces eso, el lifespan pasa a ser tu problema:

server.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from starlette.applications import Starlette
from starlette.routing import Mount

from mcp.server import MCPServer

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


app = Starlette(
    routes=[Mount("/", app=mcp.streamable_http_app())],
    lifespan=lifespan,
)
  • Mount("/", ...) junto con el path por defecto /mcp mantiene el endpoint en /mcp. Starlette prueba las rutas en orden y Mount("/") coincide con cualquier path, así que tus propias rutas van antes que él en la lista. Todo lo que quede después es inalcanzable.
  • La función lifespan entra en mcp.session_manager.run() durante toda la vida de la app anfitriona. Esta es la línea que todo el mundo olvida.
  • mcp.session_manager solo existe después de llamar a streamable_http_app(). Por eso las rutas se construyen en el ámbito del módulo y el gestor solo se toca dentro del lifespan.

La ruta Host de Starlette funciona igual: cambia Mount("/", ...) por Host("mcp.example.com", ...) para enrutar por nombre de host en lugar de por path. La regla del lifespan no cambia, y la de la seguridad del transporte tampoco. Una ruta Host("mcp.example.com", ...) solo recibe solicitudes dirigidas a ese nombre de host, pero la lista de permitidos de Host del propio transporte (Desplegar y escalar) sigue ejecutándose primero. Sin "mcp.example.com" en ella, esa ruta responde a todas con un 421.

La app anfitriona es dueña del lifespan

streamable_http_app() conecta session_manager.run() al lifespan de la Starlette que devuelve, pero el lifespan de una subaplicación montada nunca se ejecuta. Monta la app y ese lifespan integrado es código muerto. La app que esté en la cima de tu pila ASGI, sea cual sea, debe entrar en mcp.session_manager.run() en su propio lifespan.

Check

Borra la línea lifespan=lifespan y arranca el servidor. Arranca. La ruta se resuelve. Luego la primera solicitud a /mcp falla con:

RuntimeError: Task group is not initialized. Make sure to use run().

Nada arranca el gestor de sesiones salvo su run().

Dos servidores, una app

Cada MCPServer es su propia app con su propio gestor de sesiones. Monta tantos como quieras; entra en todos los gestores desde el lifespan de la app anfitriona, que es uno solo:

server.py
from collections.abc import AsyncIterator
from contextlib import AsyncExitStack, asynccontextmanager

from starlette.applications import Starlette
from starlette.routing import Mount

from mcp.server import MCPServer

notes = MCPServer("Notes")
tasks = MCPServer("Tasks")


@notes.tool()
def add_note(text: str) -> str:
    """Save a note."""
    return f"Saved: {text}"


@tasks.tool()
def add_task(title: str) -> str:
    """Create a task."""
    return f"Created: {title}"


@asynccontextmanager
async def lifespan(app: Starlette) -> AsyncIterator[None]:
    async with AsyncExitStack() as stack:
        await stack.enter_async_context(notes.session_manager.run())
        await stack.enter_async_context(tasks.session_manager.run())
        yield


app = Starlette(
    routes=[
        Mount("/notes", app=notes.streamable_http_app()),
        Mount("/tasks", app=tasks.streamable_http_app()),
    ],
    lifespan=lifespan,
)
  • AsyncExitStack entra en ambos gestores; arrancan juntos y se cierran en orden inverso.
  • Los endpoints son /notes/mcp y /tasks/mcp: el prefijo de montaje más el path por defecto.

Cambiar el path

Ese /mcp final es streamable_http_path. Ponlo en "/" y el prefijo de montaje pasa a ser el path público completo:

server.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from starlette.applications import Starlette
from starlette.routing import Mount

from mcp.server import MCPServer

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


app = Starlette(
    routes=[Mount("/notes", app=mcp.streamable_http_app(streamable_http_path="/"))],
    lifespan=lifespan,
)

Ahora los clientes se conectan a /notes, no a /notes/mcp.

CORS para clientes de navegador

Un cliente basado en navegador necesita dos permisos de tu parte: enviar sus encabezados de solicitud MCP y leer el que MCP devuelve. Ambos son configuración CORS de la app anfitriona, y la lista de permitidos de seguridad del transporte de arriba tiene que concordar con ella:

server.py
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,
)
  • allow_headers es la mitad que todo el mundo olvida. Un navegador hace un preflight de cada solicitud MCP, porque Content-Type: application/json y los encabezados de solicitud Mcp-* no están en la lista segura de CORS, y un encabezado que el preflight no concede es una solicitud que el navegador nunca envía. (allow_headers=["*"] también funciona: Starlette responde a un preflight con lo que sea que haya pedido.)
  • expose_headers=["Mcp-Session-Id"] es la mitad de lectura. Streamable HTTP devuelve el ID de sesión en ese encabezado de respuesta, y los navegadores ocultan los encabezados de respuesta a JavaScript salvo que CORS los exponga por nombre. Sin él, el cliente nunca puede hacer su segunda solicitud.
  • allow_origins es decisión tuya, no de MCP. Sé preciso y refléjalo en allowed_origins= arriba: el navegador hace cumplir CORS, pero el servidor comprueba Origin por su cuenta, y un origen en el que el transporte no confía recibe un 403 incluso tras un preflight limpio.
  • allow_methods enumera los tres métodos que usa Streamable HTTP: POST para enviar mensajes, GET para abrir el flujo de servidor a cliente, DELETE para terminar la sesión.

Rutas personalizadas

@mcp.custom_route() registra un endpoint HTTP simple en la misma app, para las cosas que todo servicio desplegado necesita y que no tienen nada que ver con MCP: una comprobación de estado, un callback de OAuth.

server.py
from starlette.requests import Request
from starlette.responses import JSONResponse, Response

from mcp.server import MCPServer

mcp = MCPServer("Notes")


@mcp.tool()
def add_note(text: str) -> str:
    """Save a note."""
    return f"Saved: {text}"


@mcp.custom_route("/health", methods=["GET"])
async def health(request: Request) -> Response:
    return JSONResponse({"status": "ok"})


app = mcp.streamable_http_app()
  • El handler es Starlette puro: una función async de Request a Response.
  • streamable_http_app() recoge todas las rutas personalizadas. app.routes es ahora /mcp y /health.
  • GET /health responde {"status": "ok"} sin rastro de MCP.

Warning

Las rutas personalizadas nunca se autentican, aunque el resto del servidor sí. Es deliberado: las comprobaciones de estado y los callbacks de OAuth tienen que ser accesibles antes de que exista ningún token. No pongas nada privado detrás de una.

Resumen

  • mcp.streamable_http_app() devuelve una app Starlette con una ruta, /mcp. Cualquier servidor ASGI puede ejecutarla.
  • Por defecto, la app responde solo a las solicitudes dirigidas a localhost, y detrás de un nombre de host real lo rechaza todo con un 421 hasta que le pases a transport_security= una lista de permitidos. Desplegar y escalar se ocupa de eso y del resto del camino a producción.
  • Mount (o Host) la mete dentro de una app Starlette o FastAPI más grande.
  • Montar desactiva el lifespan integrado. El lifespan de la app anfitriona debe entrar en mcp.session_manager.run(), o la primera solicitud falla.
  • Varios servidores en una app significa varios montajes y un solo lifespan que entra en todos los gestores de sesiones.
  • streamable_http_path="/" mueve el endpoint al propio prefijo de montaje.
  • Los clientes de navegador necesitan CORS: allow_headers para los encabezados de solicitud Mcp-*, expose_headers=["Mcp-Session-Id"] para la respuesta.
  • @mcp.custom_route() añade endpoints HTTP simples, sin autenticación, junto a /mcp.

Una vez que el servidor es accesible en una URL real, El cliente se conecta a él con esa URL en lugar de con un objeto servidor.