Додавання до наявного застосунку

Машинний переклад

Цю сторінку перекладено автоматично з англомовної документації, і основною версією є англомовна сторінка. Якщо щось читається неправильно, на сторінці Переклади пояснено, як про це повідомити.

mcp.run("streamable-http") запускає вебсервер за вас. Інколи це не те, що потрібно: MCP-сервер — лише частина більшого вебзастосунку, або ASGI-розгортання у вас уже є.

Для цього mcp.streamable_http_app() повертає застосунок Starlette.

Застосунок Starlette — це ASGI-застосунок, тож будь-що, що вміє розміщувати ASGI (uvicorn, Hypercorn, інший Starlette, FastAPI), може розмістити й ваш MCP-сервер.

Застосунок

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 — звичайний ASGI-застосунок. Передайте його будь-якому ASGI-серверу:

uvicorn server:app

Кінцева точка MCP розташована на /mcp, тож клієнт під'єднується до http://127.0.0.1:8000/mcp.

Застосунок уже містить дві речі:

Один маршрут, /mcp — кінцева точка Streamable HTTP.
Життєвий цикл (lifespan), який запускає mcp.session_manager — об'єкт, що відповідає за фонову роботу кожної активної сесії.

Запустіть застосунок окремо (uvicorn server:app) — і про жодне з них думати не доведеться.

Tip

streamable_http_app() приймає ті самі іменовані аргументи, що й mcp.run("streamable-http", ...), окрім port: порт належить тому, що обслуговує застосунок. host усе ще приймається, але тут нічого не прив'язує; що він насправді контролює, пояснює Розгортання та масштабування. Самі параметри описано на сторінці Запуск сервера.

mcp.sse_app() робить те саме для транспорту SSE, який уже витіснено новішим.

Лише localhost, доки не вкажете інше

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

Монтування

Щойно MCP-сервер стає частиною більшого застосунку, ви кладете застосунок усередину Mount. І щойно ви це робите, життєвий цикл стає вашим клопотом:

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("/", ...) разом із типовим шляхом /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", ...) отримує лише запити, адресовані цьому імені хоста, але власний список дозволених хостів транспорту (Розгортання та масштабування) усе одно спрацьовує першим. Без "mcp.example.com" у ньому цей маршрут відповідає на кожен із них кодом 421.

Життєвим циклом володіє хост-застосунок

streamable_http_app() під'єднує session_manager.run() до життєвого циклу застосунку Starlette, який повертає, але життєвий цикл змонтованого підзастосунку ніколи не виконується. Змонтуйте застосунок — і цей вбудований життєвий цикл стане мертвим кодом. Той застосунок, що стоїть на вершині вашого ASGI-стека, мусить увійти в mcp.session_manager.run() у власному життєвому циклі.

Check

Видаліть рядок lifespan=lifespan і запустіть сервер. Він запускається. Маршрут розв'язується. А потім перший запит до /mcp завершується помилкою:

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

Ніщо не запускає менеджер сесій, окрім його run().

Два сервери, один застосунок

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

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

Зміна шляху

Цей кінцевий /mcp — це streamable_http_path. Задайте йому "/" — і префікс монтування стане повним публічним шляхом:

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

Тепер клієнти під'єднуються до /notes, а не до /notes/mcp.

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

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

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

Власні маршрути

@mcp.custom_route() реєструє звичайну HTTP-кінцеву точку в тому самому застосунку — для речей, які потрібні кожному розгорнутому сервісу й не мають нічого спільного з MCP: перевірка стану, колбек 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()

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

Warning

Власні маршрути ніколи не автентифікуються, навіть коли решта сервера — так. Це навмисно: перевірки стану й колбеки OAuth мають бути досяжні ще до появи будь-якого токена. Не ховайте за ними нічого приватного.

Підсумки

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-адресою, а не через об'єкт сервера.