Добавление в существующее приложение

mcp.run("streamable-http") запускает веб-сервер за вас. Иногда это не то, что нужно: MCP-сервер — лишь часть более крупного веб-приложения, или у вас уже есть развёрнутое ASGI-приложение.

Для таких случаев mcp.streamable_http_app() возвращает приложение Starlette.

Приложение Starlette — это ASGI-приложение, поэтому разместить MCP-сервер может всё, что умеет запускать ASGI: uvicorn, Hypercorn, другое приложение Starlette, FastAPI.

Приложение

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 — обычное ASGI-приложение. Передайте его любому ASGI-серверу:

uvicorn server:app

Конечная точка MCP находится по пути /mcp, так что клиент подключается к http://127.0.0.1:8000/mcp.

В приложении уже есть две вещи:

• Один маршрут, /mcp: конечная точка Streamable HTTP.
• Жизненный цикл (lifespan), который запускает mcp.session_manager — объект, владеющий фоновой работой всех активных сессий.

Запустите приложение само по себе (uvicorn server:app) — и ни о том, ни о другом думать не придётся.

mcp.sse_app() делает то же самое для вытесненного транспорта SSE.

Только localhost, пока вы не укажете иное

По умолчанию приложение отвечает только на запросы, адресованные localhost. streamable_http_app() не может знать, за каким именем хоста его будут обслуживать, поэтому включает защиту от DNS-rebinding с самым безопасным из возможных списком разрешённых хостов; на вашей машине это ровно то, что нужно. При развёртывании за настоящим именем хоста это означает, что каждый запрос отклоняется с 421 Misdirected Request, пока вы не передадите в transport_security= список того, что действительно обслуживаете. До вашего кода дело даже не доходит. Этот список и всё остальное, что отделяет работающее приложение от настоящего имени хоста, — на странице Развёртывание и масштабирование.

Монтирование

Как только MCP-сервер становится частью более крупного приложения, вы помещаете его приложение внутрь Mount. И как только вы это делаете, жизненный цикл становится вашей заботой:

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("/", ...) вместе с путём /mcp по умолчанию оставляет конечную точку по адресу /mcp. Starlette перебирает маршруты по порядку, а Mount("/") совпадает с любым путём, поэтому ваши собственные маршруты идут в списке перед ним. Всё, что после него, недостижимо.
• Функция lifespan входит в mcp.session_manager.run() на всё время жизни хост-приложения. Именно эту строку все забывают.
• mcp.session_manager существует только после вызова streamable_http_app(). Поэтому маршруты строятся на уровне модуля, а к менеджеру обращаются только внутри жизненного цикла.

Маршрут Host из Starlette работает так же: замените Mount("/", ...) на Host("mcp.example.com", ...), чтобы маршрутизировать по имени хоста, а не по пути. Правило о жизненном цикле не меняется, как и правило о транспортной безопасности. Маршрут Host("mcp.example.com", ...) получает только запросы, адресованные этому имени хоста, но собственный список разрешённых значений Host у транспорта (Развёртывание и масштабирование) всё равно проверяется первым. Если в нём нет "mcp.example.com", этот маршрут отвечает на каждый такой запрос кодом 421.

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

Два сервера, одно приложение

Каждый MCPServer — отдельное приложение со своим менеджером сессий. Монтируйте сколько угодно; входите в каждый менеджер из одного жизненного цикла хост-приложения:

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 входит в оба менеджера; они запускаются вместе и завершаются в обратном порядке.
• Конечные точки — /notes/mcp и /tasks/mcp: префикс монтирования плюс путь по умолчанию.

Изменение пути

Завершающий /mcp — это streamable_http_path. Задайте ему значение "/", и префикс монтирования станет полным публичным путём:

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,
)

Теперь клиенты подключаются к /notes, а не к /notes/mcp.

CORS для браузерных клиентов

Браузерному клиенту нужны от вас два разрешения: отправлять свои заголовки MCP-запроса и читать тот заголовок, что MCP присылает в ответ. И то и другое — настройка CORS в хост-приложении, и список разрешённых хостов транспортной безопасности, описанный выше, должен с ней согласовываться:

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 — та половина, которую все забывают. Браузер выполняет предварительный запрос (preflight) перед каждым MCP-запросом, потому что Content-Type: application/json и заголовки запроса Mcp-* не входят в безопасный список CORS, а заголовок, не разрешённый предварительным запросом, — это запрос, который браузер никогда не отправит. (allow_headers=["*"] тоже работает: Starlette отвечает на предварительный запрос тем, что тот запросил.)
• expose_headers=["Mcp-Session-Id"] — половина, отвечающая за чтение. Streamable HTTP возвращает идентификатор сессии в этом заголовке ответа, а браузеры скрывают заголовки ответа от JavaScript, если CORS не раскрывает их поимённо. Без этого клиент никогда не сможет сделать второй запрос.
• allow_origins — ваше решение, а не MCP. Будьте точны и продублируйте его в allowed_origins= выше: CORS обеспечивает браузер, но сервер сам проверяет Origin, и источник, которому транспорт не доверяет, получает 403 даже после успешного предварительного запроса.
• allow_methods перечисляет три метода, которые использует Streamable HTTP: POST для отправки сообщений, GET для открытия потока от сервера к клиенту, DELETE для завершения сессии.

Пользовательские маршруты

@mcp.custom_route() регистрирует обычную HTTP-точку в том же приложении — для вещей, которые нужны каждому развёрнутому сервису и не имеют отношения к MCP: проверка работоспособности, колбэк OAuth.

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()

• Обработчик — обычный Starlette: async-функция из Request в Response.
• streamable_http_app() подхватывает все пользовательские маршруты. Теперь app.routes — это /mcp и /health.
• GET /health отвечает {"status": "ok"} без всякого MCP.

Итоги

• mcp.streamable_http_app() возвращает приложение Starlette с одним маршрутом, /mcp. Запустить его может любой ASGI-сервер.
• По умолчанию приложение отвечает только на запросы, адресованные localhost, а за настоящим именем хоста отклоняет всё кодом 421, пока вы не передадите список разрешённых хостов в transport_security=. За это и за остальной путь к продакшену отвечает страница Развёртывание и масштабирование.
• Mount (или Host) помещает его внутрь более крупного приложения Starlette или FastAPI.
• Монтирование отключает встроенный жизненный цикл. Жизненный цикл хост-приложения должен войти в mcp.session_manager.run(), иначе первый запрос завершится ошибкой.
• Несколько серверов в одном приложении — это несколько монтирований и один жизненный цикл, который входит в каждый менеджер сессий.
• streamable_http_path="/" переносит конечную точку на сам префикс монтирования.
• Браузерным клиентам нужен CORS: allow_headers для заголовков запроса Mcp-*, expose_headers=["Mcp-Session-Id"] для ответа.
• @mcp.custom_route() добавляет обычные HTTP-точки без аутентификации рядом с /mcp.

Когда сервер доступен по настоящему URL, Клиент подключается к нему по этому URL вместо объекта сервера.