Ana içeriğe geç

Mevcut bir uygulamaya ekleme

Makine çevirisi

Bu sayfa İngilizce dokümantasyondan otomatik olarak çevrildi; esas alınması gereken sürüm İngilizce sayfadır. Yanlış görünen bir şey varsa, nasıl bildireceğinizi Çeviriler sayfası açıklar.

mcp.run("streamable-http") sizin için bir web sunucusu başlatır. Bazen bunu istemezsiniz: MCP sunucunuz daha büyük bir web uygulamasının bir parçasıdır ya da zaten bir ASGI dağıtımınız vardır.

Bunun için mcp.streamable_http_app() bir Starlette uygulaması döndürür.

Starlette uygulaması bir ASGI uygulamasıdır; dolayısıyla ASGI barındırabilen her şey (uvicorn, Hypercorn, başka bir Starlette, FastAPI) MCP sunucunuzu da barındırabilir.

Uygulama

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 sıradan bir ASGI uygulamasıdır. Herhangi bir ASGI sunucusuna verin:

uvicorn server:app

MCP endpoint'i /mcp yolundadır; yani istemci http://127.0.0.1:8000/mcp adresine bağlanır.

Uygulama hâlihazırda iki şey taşır:

  • Tek bir rota, /mcp: Streamable HTTP endpoint'i.
  • mcp.session_manager'ı başlatan bir lifespan (yaşam döngüsü); bu nesne, canlı her oturumun arka plan işlerinin sahibidir.

Uygulamayı tek başına çalıştırın (uvicorn server:app), ikisini de hiç düşünmeniz gerekmez.

Tip

streamable_http_app(), mcp.run("streamable-http", ...) ile aynı anahtar sözcük argümanlarını alır; port hariç: port, uygulamayı sunan şeye aittir. host hâlâ kabul edilir ama burada hiçbir şeye bağlanmaz; gerçekte neyi denetlediğini Dağıtım ve ölçekleme açıklar. Seçeneklerin kendisi Sunucunuzu çalıştırma sayfasında.

mcp.sse_app() aynısını, yerini yenisine bırakmış SSE aktarımı için yapar.

Siz aksini söyleyene kadar yalnızca localhost

Varsayılan olarak uygulama yalnızca localhost'a gönderilen istekleri yanıtlar. streamable_http_app() hangi ana bilgisayar adının arkasında sunulacağını bilemez; bu yüzden DNS rebinding korumasını olabilecek en güvenli izin listesiyle etkinleştirir. Kendi makinenizde bu tam olarak doğru olandır. Gerçek bir ana bilgisayar adının arkasına dağıtıldığında ise, transport_security= parametresine gerçekte sunduğunuz adların izin listesini geçirene kadar her istek 421 Misdirected Request ile reddedilir demektir. Sizin yazdığınız hiçbir şeye önce danışılmaz bile. Bu izin listesi ve çalışan bir uygulama ile gerçek bir ana bilgisayar adı arasındaki diğer her şey Dağıtım ve ölçekleme sayfasında.

Mount etme

MCP sunucusu daha büyük bir uygulamanın parçası olduğu anda uygulamayı bir Mount içine koyarsınız. Bunu yaptığınız anda da lifespan sizin sorununuz olur:

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("/", ...) ile varsayılan /mcp yolu birlikte endpoint'i /mcp yolunda tutar. Starlette rotaları sırayla dener ve Mount("/") her yolla eşleşir; bu yüzden kendi rotalarınız listede ondan önce gelir. Ondan sonraki hiçbir şeye ulaşılamaz.
  • lifespan fonksiyonu, ana uygulamanın ömrü boyunca mcp.session_manager.run() içine girer. Herkesin unuttuğu satır budur.
  • mcp.session_manager ancak streamable_http_app() çağrıldıktan sonra var olur. Rotaların modül düzeyinde kurulmasının ve yöneticiye yalnızca lifespan içinde dokunulmasının nedeni budur.

Starlette'in Host rotası aynı şekilde çalışır: yola göre değil ana bilgisayar adına göre yönlendirmek için Mount("/", ...) yerine Host("mcp.example.com", ...) koyun. Lifespan kuralı değişmez, aktarım güvenliği kuralı da. Host("mcp.example.com", ...) rotası yalnızca o ana bilgisayar adına gönderilen istekleri alır, ancak aktarımın kendi Host izin listesi (Dağıtım ve ölçekleme) yine de önce çalışır. Listede "mcp.example.com" yoksa bu rota o isteklerin her birini 421 ile yanıtlar.

Ana uygulama lifespan'in sahibidir

streamable_http_app(), session_manager.run()'ı döndürdüğü Starlette'in lifespan'ine bağlar; ancak mount edilmiş bir alt uygulamanın lifespan'i hiçbir zaman çalışmaz. Uygulamayı mount edin, o yerleşik lifespan ölü kod olur. ASGI yığınınızın en üstünde hangi uygulama duruyorsa, kendi lifespan'inde mcp.session_manager.run() içine girmelidir.

Check

lifespan=lifespan satırını silin ve sunucuyu başlatın. Başlar. Rota çözülür. Sonra /mcp yoluna gelen ilk istek şu hatayla başarısız olur:

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

Oturum yöneticisini kendi run()'ından başka hiçbir şey başlatmaz.

İki sunucu, tek uygulama

Her MCPServer, kendi oturum yöneticisi olan ayrı bir uygulamadır. İstediğiniz kadarını mount edin; her yöneticiye tek ana lifespan'den girin:

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 iki yöneticiye de girer; birlikte başlar, ters sırada kapanırlar.
  • Endpoint'ler /notes/mcp ve /tasks/mcp: mount öneki artı varsayılan yol.

Yolu değiştirme

Sondaki o /mcp, streamable_http_path değeridir. Bunu "/" yapın, mount öneki genel yolun tamamı olur:

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

Artık istemciler /notes/mcp yoluna değil /notes yoluna bağlanır.

Tarayıcı istemcileri için CORS

Tarayıcı tabanlı bir istemcinin sizden iki izne ihtiyacı vardır: MCP istek başlıklarını göndermek ve MCP'nin geri gönderdiği başlığı okumak. İkisi de ana uygulamadaki CORS yapılandırmasıdır ve yukarıdaki aktarım güvenliği izin listesinin bununla uyuşması gerekir:

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 herkesin unuttuğu yarıdır. Tarayıcı her MCP isteği için preflight yapar; çünkü Content-Type: application/json ve Mcp-* istek başlıkları CORS güvenli listesinde değildir ve preflight'ın izin vermediği bir başlık, tarayıcının asla göndermediği bir istek demektir. (allow_headers=["*"] da çalışır: Starlette bir preflight'ı ne istediyse onunla yanıtlar.)
  • expose_headers=["Mcp-Session-Id"] okuma yarısıdır. Streamable HTTP oturum kimliğini bu yanıt başlığında döndürür ve tarayıcılar, CORS adlarıyla açığa çıkarmadıkça yanıt başlıklarını JavaScript'ten gizler. Bu olmadan istemci ikinci isteğini asla yapamaz.
  • allow_origins MCP'nin değil sizin kararınızdır. Kesin olun ve yukarıdaki allowed_origins= ile birebir eşleştirin: CORS'u tarayıcı uygular, ama sunucu Origin'i kendisi de denetler ve aktarımın güvenmediği bir origin, temiz bir preflight'tan sonra bile 403 alır.
  • allow_methods Streamable HTTP'nin kullandığı üç yöntemi listeler: ileti göndermek için POST, sunucudan istemciye akışı açmak için GET, oturumu sonlandırmak için DELETE.

Özel rotalar

@mcp.custom_route() aynı uygulamada düz bir HTTP endpoint'i kaydeder; dağıtılan her servisin ihtiyaç duyduğu ama MCP ile hiçbir ilgisi olmayan şeyler için: sağlık denetimi, OAuth callback'i.

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()
  • İşleyici düz Starlette'tir: Request'ten Response'a bir async fonksiyon.
  • streamable_http_app() her özel rotayı alır. app.routes artık /mcp ve /health.
  • GET /health, ortada hiç MCP olmadan {"status": "ok"} yanıtını verir.

Warning

Özel rotalar, sunucunun geri kalanı doğrulansa bile hiçbir zaman kimlik doğrulamasından geçmez. Bu kasıtlıdır: sağlık denetimleri ve OAuth callback'leri herhangi bir token var olmadan önce erişilebilir olmak zorundadır. Bunların arkasına özel hiçbir şey koymayın.

Özet

  • mcp.streamable_http_app() tek rotası /mcp olan bir Starlette uygulaması döndürür. Herhangi bir ASGI sunucusu onu çalıştırabilir.
  • Varsayılan olarak uygulama yalnızca localhost'a gönderilen istekleri yanıtlar; gerçek bir ana bilgisayar adının arkasında ise transport_security= parametresine bir izin listesi geçirene kadar her şeyi 421 ile reddeder. Bu konu ve üretime giden yolun geri kalanı Dağıtım ve ölçekleme sayfasında.
  • Mount (veya Host) onu daha büyük bir Starlette ya da FastAPI uygulamasının içine koyar.
  • Mount etmek yerleşik lifespan'i devre dışı bırakır. Ana uygulamanın lifespan'i mcp.session_manager.run() içine girmelidir, yoksa ilk istek başarısız olur.
  • Tek uygulamada birden fazla sunucu, birden fazla mount ve her oturum yöneticisine giren tek bir lifespan demektir.
  • streamable_http_path="/" endpoint'i mount önekinin kendisine taşır.
  • Tarayıcı istemcilerinin CORS'a ihtiyacı vardır: Mcp-* istek başlıkları için allow_headers, yanıt için expose_headers=["Mcp-Session-Id"].
  • @mcp.custom_route(), /mcp'nin yanına düz, kimlik doğrulaması olmayan HTTP endpoint'leri ekler.

Sunucu gerçek bir URL'den erişilebilir olduğunda İstemci ona bir sunucu nesnesi yerine o URL ile bağlanır.