Einfache HTTP-MCP-Server-Implementierung

Dieses Projekt bietet eine leichtgewichtige Server-Implementierung für das Model Context Protocol (MCP) über HTTP. Es ermöglicht, Python-Funktionen als Tools und Prompts bereitzustellen, die über eine JSON-RPC-Schnittstelle remote entdeckt und ausgeführt werden können. Es ist für die Verwendung mit einer Starlette- oder FastAPI-Anwendung gedacht (siehe Demo).

Inhaltsverzeichnis

• Funktionen

• Installation

• Serverarchitektur

• Protokollversion

  • Die 2026-07-28-Anfrageform

  • Caching-Hinweise

  • Ursprungsvalidierung

  • Spiegelung von Tool-Parametern in Header

• Tools

  • Einfaches Tool-Beispiel

  • Tools ohne Argumente

  • Tools mit Fehlerbehandlung

  • Tools mit Autorisierungsbereichen

• Serverzustandsverwaltung

• Zugriff auf Anfragen

• Prompts

  • Einfaches Prompt-Beispiel

  • Prompts ohne Argumente

  • Prompts mit Lebenszyklus-Zustand

  • Prompts mit Autorisierungsbereichen

• STDIO-Transport

• Authentifizierung und Autorisierung

• OAuth-2.1-Autorisierung (auth_mcp)

• API-Referenz

• Sicherheitsflächen nach Endpunkt

• Lizenz

Related MCP server: remote-mcp

Funktionen

• MCP-Protokollkonform: Implementiert die MCP-Spezifikation für die Entdeckung und Ausführung von Tools und Prompts. Keine Unterstützung für Benachrichtigungen.

• Eine Protokollrevision: Spricht ausschließlich die zustandslose Revision 2026-07-28 – server/discover, _meta pro Anfrage, kein Handshake, keine Sitzung. Ein einziger Dispatch-Pfad bedeutet, dass eine Anfrage keine schwächere Behandlung auswählen kann, indem sie eine ältere Revision deklariert.

• HTTP- und STDIO-Transport: Verwendet HTTP (POST-Anfragen) oder STDIO für die Kommunikation.

• Async-Unterstützung: Basiert auf Starlette oder FastAPI für die asynchrone Anfrageverarbeitung.

• Typsicher: Nutzt Pydantic für robuste Datenvalidierung und -serialisierung.

• Serverzustandsverwaltung: Zugriff auf gemeinsamen Zustand über den Lebenszyklus-Kontext mithilfe der get_state_key-Methode.

• Zugriff auf Anfragen: Zugriff auf das eingehende Anfrageobjekt aus Ihren Tools und Prompts.

• Autorisierungsbereiche: Unterstützung für bereichsbasierte Autorisierung mithilfe des Starlette-Authentifizierungssystems.

• Fehlerbehandlung: Tools können optional Fehlermeldungen zurückgeben, anstatt Ausnahmen auszulösen.

• OAuth-2.1-Autorisierung: Optionales auth_mcp-Paket mit Bearer-Token-Validierung, Protected Resource Metadata (RFC 9728) und WWW-Authenticate-Fehlerantworten. Installation mit pip install http-mcp[auth].

Serverarchitektur

Die Bibliothek bietet eine einzelne MCPServer-Klasse, die den Lebenszyklus zur Verwaltung des gemeinsamen Zustands über den gesamten Anwendungslebenszyklus nutzt.

MCPServer

Der MCPServer ist für die Verwendung mit dem Lebenszyklus-System von Starlette zur Verwaltung des gemeinsamen Serverzustands konzipiert.

Wesentliche Merkmale:

• Lebenszyklusbasiert: Verwendet Starlette-Lebenszyklusereignisse zur Initialisierung und Verwaltung des gemeinsamen Serverzustands.

• Anwendungsebene: Der Zustand bleibt über den gesamten Anwendungslebenszyklus bestehen, nicht pro Anfrage.

• Flexibel: Kann mit jeder benutzerdefinierten Kontextklasse verwendet werden, die im Lebenszyklus-Zustand gespeichert ist.

Konstruktorparameter:

• name (str): Der Name Ihres MCP-Servers.

• version (str): Die Version Ihres MCP-Servers.

• tools (tuple[Tool, ...]): Tupel der bereitzustellenden Tools (Standard: leeres Tupel).

• prompts (tuple[Prompt, ...]): Tupel der bereitzustellenden Prompts (Standard: leeres Tupel).

• instructions (str | None): Optionale Anweisungen für KI-Assistenten zur Verwendung dieses Servers.

• cache_ttl_ms (int): Frischehinweis in Millisekunden, der mit tools/list, prompts/list und server/discover-Ergebnissen gesendet wird (Standard: 300000). Verwenden Sie 0, um Clients mitzuteilen, niemals zu cachen. Siehe Caching-Hinweise.

• cache_scope ("public" | "private" | None): Gibt an, ob gemeinsame Caches diese Ergebnisse über Autorisierungskontexte hinweg wiederverwenden dürfen. Wird automatisch abgeleitet, wenn weggelassen. Siehe Caching-Hinweise.

• allowed_origins (tuple[str, ...]): Ursprünge, die der HTTP-Transport akzeptiert (Standard: leer, d. h. die Prüfung ist deaktiviert). Siehe Ursprungsvalidierung.

• require_origin (bool): Gibt an, ob eine Anfrage ohne Origin-Header abgelehnt wird, wenn allowed_origins gesetzt ist (Standard: False). Siehe Ursprungsvalidierung.

Beispielverwendung:

import contextlib
from collections.abc import AsyncIterator
from typing import TypedDict
from dataclasses import dataclass, field
from starlette.applications import Starlette
from http_mcp.server import MCPServer

@dataclass
class Context:
    call_count: int = 0
    user_preferences: dict = field(default_factory=dict)

class State(TypedDict):
    context: Context

@contextlib.asynccontextmanager
async def lifespan(_app: Starlette) -> AsyncIterator[State]:
    yield {"context": Context()}

mcp_server = MCPServer(
    name="my-server",
    version="1.0.0",
    tools=my_tools,
    prompts=my_prompts,
    instructions="Optional instructions for AI assistants on how to use this server"
)

app = Starlette(lifespan=lifespan)
app.mount("/mcp", mcp_server.app)

Protokollversion

Der Server implementiert genau eine Protokollrevision, 2026-07-28, und jede Anfrage durchläuft denselben Pfad. Es gibt keine Versionsverhandlung und kein zweites Regelwerk, in das eine Anfrage wechseln kann.

Breaking Change in 0.17.0. Die Unterstützung für die sitzungsbasierten Revisionen 2025-11-25, 2025-06-18 und 2025-03-26 wurde entfernt, ebenso wie initialize, notifications/initialized und ping. Ein Client, der nur diese Revisionen spricht, kann nicht mehr mit diesem Server kommunizieren. Das Bedienen einer einzigen Revision macht auch die unten genannten Anfrage-Metadaten-Header vertrauenswürdig: Während zwei Epochen koexistierten, konnte eine Anfrage die Header-Prüfungen umgehen, indem sie die ältere deklarierte, sodass ein Intermediär, der auf Mcp-Method routet, vom Server, der auf den Body reagiert, desynchronisiert werden konnte.

Breaking Change in 0.18.0. ServerInterface.get_tool_input_schema akzeptiert jetzt die Request, sodass Autorisierungsbereiche vor dem Dispatch berücksichtigt werden; Implementierungen der Schnittstelle müssen aktualisiert werden, während MCPServer-Benutzer nicht betroffen sind. Gespiegelte Mcp-Param-*-Werte werden textuell statt numerisch verglichen, sodass ein Header mit 3.0 für "replicas": 3 jetzt -32020 erhält. Jede notifications/*-Methode gibt 404 zurück, da die Revision keine definiert.

Die 2026-07-28-Anfrageform

Die Revision hat kein Sitzungskonzept. In der Praxis:

• Kein Handshake. Jede Anfrage wiederholt ihre Protokollversion und die Client-Fähigkeiten in _meta:

  {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
      "name": "get_weather",
      "arguments": { "location": "Seattle, WA" },
      "_meta": {
        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientInfo": { "name": "ExampleClient", "version": "1.0.0" },
        "io.modelcontextprotocol/clientCapabilities": {}
      }
    }
  }

  protocolVersion und clientCapabilities sind erforderlich; das Weglassen eines von beiden führt zu -32602 und HTTP 400. Jede andere Version führt zu -32022, dessen data.supported die eine Revision auflistet, die dieser Server spricht.

• server/discover ersetzt initialize für die Fähigkeitserkennung. Es meldet die unterstützte Version, Fähigkeiten, Anweisungen und Serveridentität in einem Aufruf und wird ohne vorherige Anfrage beantwortet:

  {
    "resultType": "complete",
    "supportedVersions": ["2026-07-28"],
    "capabilities": { "tools": { "listChanged": false }, "prompts": { "listChanged": false } },
    "_meta": { "io.modelcontextprotocol/serverInfo": { "name": "my-server", "version": "1.0.0" } },
    "ttlMs": 300000,
    "cacheScope": "public"
  }

• Jedes Ergebnis trägt resultType: "complete" und einen _meta-Block, der den Server benennt.

• initialize, notifications/initialized, ping und logging/setLevel existieren nicht, ebenso wie die Sitzungs- und SSE-Wiederaufnahme-Mechanismen. Sie geben -32601 mit HTTP 404 zurück. JSON-RPC-Benachrichtigungen – eine notifications/*-Nachricht ohne id – erhalten weiterhin 202 Accepted und keinen Body, da JSON-RPC Antworten darauf verbietet.

• Erforderliche Anfrage-Header. Jeder POST muss MCP-Protocol-Version und Mcp-Method senden, zusätzlich Mcp-Name bei tools/call und prompts/get. Jeder muss mit dem entsprechenden Body-Wert übereinstimmen, sonst wird die Anfrage mit -32020 (HeaderMismatch) und HTTP 400 abgelehnt – dies verhindert, dass ein Proxy auf einen Wert routet, während der Server auf einen anderen reagiert. Werte, die nicht als einfaches ASCII ausgedrückt werden können, verwenden die =?base64?...?=-Hülle, die der Server vor dem Vergleich dekodiert.

• Unbekannte Tools und Prompts melden -32602, was die Tools- und Prompts-Spezifikationen vorschreiben. -32002 wurde durch diese Revision zurückgezogen.

• Mcp-Session-Id und Last-Event-ID werden ignoriert, und GET/DELETE auf dem MCP-Endpunkt geben 405 Method Not Allowed zurück.

Mehrstufige Round-Trip-Anfragen (Elicitation, Sampling, Roots) und subscriptions/listen sind nicht implementiert: Dieser Server stellt keine clientabhängigen Funktionen bereit und deklariert listChanged: false, sodass keines von beiden auf ihn zutrifft.

Caching-Hinweise

tools/list, prompts/list und server/discover-Ergebnisse tragen ttlMs und cacheScope, damit Clients vermeiden können, eine Liste erneut abzurufen, die sich nicht geändert hat:

mcp_server = MCPServer(
    name="my-server",
    version="1.0.0",
    tools=my_tools,
    cache_ttl_ms=300_000,   # clients may treat the list as fresh for 5 minutes
    cache_scope="public",   # shared caches may serve it to any caller
)

Tools und Prompts sind bei der Konstruktion von MCPServer festgelegt, sodass ttlMs tatsächlich begrenzt, wie lange ein Client einen erneuten Deployment verpassen darf, und nicht, wie stabil die Daten sind. Setzen Sie es auf 0, um Clients zu bitten, niemals zu cachen.

cache_scope wird abgeleitet, wenn Sie es weglassen: "private", wenn ein Tool oder Prompt bereichsbeschränkt ist – die Liste variiert dann pro Aufrufer, sodass ein gemeinsamer Cache sie nicht über Autorisierungskontexte hinweg wiederverwenden darf – und andernfalls "public". Überschreiben Sie es, wenn Ihr Deployment es besser weiß. Beachten Sie, dass cacheScope nur das Caching regelt; es ist niemals ein Ersatz für die bereichsbezogenen Prüfungen pro Tool.

Ursprungsvalidierung

Browser fügen einen Origin-Header hinzu, was es einem Server ermöglicht, Anfragen abzulehnen, die durch DNS-Rebinding eingeschleust wurden. Die Prüfung ist standardmäßig deaktiviert, damit bestehende Deployments weiter funktionieren; aktivieren Sie sie überall dort, wo der Endpunkt von einem Browser aus erreichbar ist:

mcp_server = MCPServer(
    name="my-server",
    version="1.0.0",
    tools=my_tools,
    allowed_origins=("https://app.example.com",),
)

Eine Anfrage, deren Origin vorhanden und nicht auf der Liste ist, erhält 403 Forbidden. Anfragen ohne Origin – normale Nicht-Browser-Clients – sind standardmäßig nicht betroffen, da Browser bei einem POST immer Origin senden und das Rebinding-Bedrohungsmodell keine Clients abdeckt, die keine Browser sind.

Wenn der Endpunkt nur Browserverkehr bedienen soll, fügen Sie require_origin hinzu, um auch eine Anfrage abzulehnen, die den Header weglässt, wodurch die Whitelist verbindlich statt beratend wird:

mcp_server = MCPServer(
    name="my-server",
    version="1.0.0",
    tools=my_tools,
    allowed_origins=("https://app.example.com",),
    require_origin=True,
)

require_origin bewirkt allein nichts – es verschärft nur eine bereits konfigurierte Whitelist. Binden Sie bei lokaler Ausführung auch an 127.0.0.1 statt an 0.0.0.0.

Spiegelung von Tool-Parametern in Header

Ein Tool kann Clients bitten, bestimmte Argumentwerte in Mcp-Param-*-Header zu kopieren, damit Proxys darauf routen oder raten können, ohne den Body zu parsen. Kommentieren Sie das Feld mit x-mcp-header:

from pydantic import BaseModel, Field

class ExecuteSQLInput(BaseModel):
    region: str = Field(
        description="The region to execute the query in",
        json_schema_extra={"x-mcp-header": "Region"},
    )
    query: str = Field(description="The SQL query to execute")

Ein konformer Client sendet dann Mcp-Param-Region: us-west1 zusammen mit dem Aufruf, und der Server überprüft es gegen den Body – und lehnt die Anfrage mit -32020 ab, wenn der Header fehlt, dem Argument widerspricht oder gesendet wird, wenn das Argument fehlt. Mcp-Param-*-Header, die keine Annotation beansprucht, werden ignoriert, da Intermediäre erwartungsgemäß unbekannte unverändert weiterleiten.

Der Vergleich ist textuell, gegen den Wert, wie JSON ihn schreibt: Für "replicas": 3 muss der Header exakt 3 lauten, nicht 3.0, +3 oder 3. Numerische Umwandlung würde diese als gleich betrachten, während ein Intermediär, der auf den rohen Header-String routet, etwas anderes sieht – genau die Desynchronisation, die die Spiegelung verhindern soll.

Nur string, integer und boolean-Felder, die über eine einfache Kette von Objekteigenschaften erreichbar sind, können annotiert werden, und keine zwei Felder dürfen denselben Header-Namen beanspruchen – eine Kollision wird bei der Konstruktion des Servers abgelehnt, da das Beibehalten einer der beiden Annotationen die andere stillschweigend unerzwungen ließe. Annotieren Sie keine sensiblen Werte: Header-Inhalte sind für jeden Intermediär auf dem Pfad sichtbar.

Tools

Tools sind die Funktionen, die vom Client aufgerufen werden können.

Einfaches Tool-Beispiel

1. Definieren Sie die Argumente und die Ausgabe für die Tools:

# app/tools/models.py
from pydantic import BaseModel, Field

class GreetInput(BaseModel):
    question: str = Field(description="The question to answer")

class GreetOutput(BaseModel):
    answer: str = Field(description="The answer to the question")

# Note: the description on Field will be passed when listing the tools.
# Having a description is optional, but it's recommended to provide one.

2. Definieren Sie die Tools:

# app/tools/tools.py
from http_mcp.types import Arguments

from app.tools.models import GreetInput, GreetOutput

def greet(args: Arguments[GreetInput]) -> GreetOutput:
    return GreetOutput(answer=f"Hello, {args.inputs.question}!")

# app/tools/__init__.py

from http_mcp.types import Tool
from app.tools.models import GreetInput, GreetOutput
from app.tools.tools import greet

TOOLS = (
    Tool(
        func=greet,
        inputs=GreetInput,
        output=GreetOutput,
    ),
)

__all__ = ["TOOLS"]

3. Instanziieren Sie den Server:

# app/main.py
from starlette.applications import Starlette
from http_mcp.server import MCPServer
from app.tools import TOOLS

mcp_server = MCPServer(tools=TOOLS, name="test", version="1.0.0")

app = Starlette()
app.mount(
    "/mcp",
    mcp_server.app,
)

Tools ohne Argumente

Sie können Tools definieren, die keine Eingabeargumente erfordern:

from datetime import UTC, datetime
from pydantic import BaseModel, Field
from http_mcp.types import Tool

class GetTimeOutput(BaseModel):
    time: str = Field(description="The current time")

async def get_time() -> GetTimeOutput:
    """Get the current time."""
    return GetTimeOutput(time=datetime.now(UTC).strftime("%H:%M:%S"))

TOOLS = (
    Tool(
        func=get_time,
        inputs=type(None),  # No arguments required
        output=GetTimeOutput,
    ),
)

Alternativ können Sie die NoArguments-Klasse für bessere Klarheit verwenden:

from http_mcp.types import Arguments, NoArguments, Tool

class SimpleOutput(BaseModel):
    success: bool = Field(description="Whether the operation was successful")

def simple_tool(args: Arguments[NoArguments]) -> SimpleOutput:
    """A simple tool with no arguments."""
    # You can still access request and state
    context = args.get_state_key("context", Context)
    return SimpleOutput(success=True)

TOOLS = (
    Tool(
        func=simple_tool,
        inputs=NoArguments,
        output=SimpleOutput,
    ),
)

Tools mit Fehlerbehandlung

Tools können optional Fehlermeldungen zurückgeben, anstatt Ausnahmen auszulösen:

from pydantic import BaseModel, Field
from http_mcp.types import Arguments, Tool
from http_mcp.exceptions import ToolInvocationError

class RiskyToolInput(BaseModel):
    value: int = Field(description="An integer value")

class RiskyToolOutput(BaseModel):
    result: str = Field(description="The result of the operation")

def risky_tool(args: Arguments[RiskyToolInput]) -> RiskyToolOutput:
    """A tool that might fail."""
    if args.inputs.value < 0:
        raise ToolInvocationError("risky_tool", "Value must be positive")
    return RiskyToolOutput(result=f"Success: {args.inputs.value}")

TOOLS = (
    Tool(
        func=risky_tool,
        inputs=RiskyToolInput,
        output=RiskyToolOutput,
        return_error_message=True,  # Return ErrorMessage instead of raising
    ),
)

Wenn return_error_message=True gesetzt ist, gibt das Tool ein ErrorMessage-Modell mit den Fehlerdetails zurück, anstatt eine ToolInvocationError auszulösen.

Tools mit Autorisierungsbereichen

Sie können den Tool-Zugriff basierend auf Authentifizierungsbereichen einschränken:

from http_mcp.exceptions import ToolInvocationError
from http_mcp.types import Arguments, NoArguments, Tool
from starlette.authentication import has_required_scope

class SecureOutput(BaseModel):
    message: str = Field(description="A secure message")

def private_tool(args: Arguments[NoArguments]) -> SecureOutput:
    """A tool that requires authentication."""
    if not has_required_scope(args.request, ("private",)):
        raise ToolInvocationError("private_tool", "Insufficient scope")
    return SecureOutput(message="This is private data")

def admin_tool(args: Arguments[NoArguments]) -> SecureOutput:
    """A tool that requires admin or superuser scope."""
    if not has_required_scope(args.request, ("admin", "superuser")):
        raise ToolInvocationError("admin_tool", "Insufficient scope")
    return SecureOutput(message="This is admin data")

TOOLS = (
    Tool(
        func=private_tool,
        inputs=NoArguments,
        output=SecureOutput,
        scopes=("private",),  # Only accessible with 'private' scope
    ),
    Tool(
        func=admin_tool,
        inputs=NoArguments,
        output=SecureOutput,
        scopes=("admin", "superuser"),  # Accessible with either scope
    ),
)

Hinweis: Sie müssen in Ihrer Starlette-App eine Authentifizierungs-Middleware einrichten, damit Scopes ordnungsgemäß funktionieren. Das Feld scopes auf Tool ist das primäre Autorisierungs-Gate — das Framework filtert Tools vor dem Aufruf nach Scope. Die raise ToolInvocationError(...)-Aufrufe in den obigen Tool-Funktionen sind optionale Defense-in-Depth-Prüfungen, die eine ordnungsgemäße Fehlerantwort an den Client zurückgeben, anstatt stillschweigend zu scheitern.

Server-Zustandsverwaltung

Der Server verwendet das Lifespan-System von Starlette, um gemeinsamen Zustand über den gesamten Anwendungslebenszyklus zu verwalten. Der Zustand wird beim Start der Anwendung initialisiert und bleibt bis zum Herunterfahren erhalten. Der Zugriff auf den Kontext erfolgt über die Methode get_state_key auf dem Arguments-Objekt.

Dies ist nützlich, um Ressourcen wie Datenbank-Verbindungspools, HTTP-Clients, Caches oder beliebigen Anwendungszustand über mehrere Tools hinweg gemeinsam zu nutzen.

Datenbank-Verbindungspool

Das häufigste Muster — einen Verbindungspool beim Start initialisieren, ihn über alle Tools hinweg gemeinsam nutzen und beim Herunterfahren schließen:

# app/context.py
from dataclasses import dataclass
import asyncpg

@dataclass
class AppContext:
    db: asyncpg.Pool

# app/main.py
import contextlib
import os
from collections.abc import AsyncIterator
from typing import TypedDict
import asyncpg
from starlette.applications import Starlette
from http_mcp.server import MCPServer
from app.context import AppContext

class State(TypedDict):
    ctx: AppContext

@contextlib.asynccontextmanager
async def lifespan(_app: Starlette) -> AsyncIterator[State]:
    pool = await asyncpg.create_pool(os.environ["DATABASE_URL"])
    yield {"ctx": AppContext(db=pool)}
    await pool.close()

mcp_server = MCPServer(tools=TOOLS, name="my-server", version="1.0.0")

app = Starlette(lifespan=lifespan)
app.mount("/mcp", mcp_server.app)

# app/tools.py
from pydantic import BaseModel, Field
from http_mcp.types import Arguments
from app.context import AppContext

class GetUserInput(BaseModel):
    user_id: int = Field(description="The user ID to look up")

class GetUserOutput(BaseModel):
    name: str = Field(description="The user's name")
    email: str = Field(description="The user's email")

async def get_user(args: Arguments[GetUserInput]) -> GetUserOutput:
    """Look up a user by ID."""
    ctx = args.get_state_key("ctx", AppContext)
    row = await ctx.db.fetchrow(
        "SELECT name, email FROM users WHERE id = $1",
        args.inputs.user_id,
    )
    return GetUserOutput(name=row["name"], email=row["email"])

Gemeinsamer HTTP-Client

Nutzen Sie einen einzelnen httpx.AsyncClient über mehrere Tools hinweg gemeinsam, um Verbindungen wiederzuverwenden und Basis-URLs, Header oder Timeouts einmalig zu konfigurieren:

# app/context.py
from dataclasses import dataclass
import httpx

@dataclass
class AppContext:
    http_client: httpx.AsyncClient

# app/main.py
import contextlib
from collections.abc import AsyncIterator
from typing import TypedDict
import httpx
from starlette.applications import Starlette
from http_mcp.server import MCPServer
from app.context import AppContext

class State(TypedDict):
    ctx: AppContext

@contextlib.asynccontextmanager
async def lifespan(_app: Starlette) -> AsyncIterator[State]:
    async with httpx.AsyncClient(
        base_url="https://api.example.com",
        headers={"Authorization": "Bearer <token>"},
    ) as client:
        yield {"ctx": AppContext(http_client=client)}

mcp_server = MCPServer(tools=TOOLS, name="my-server", version="1.0.0")

app = Starlette(lifespan=lifespan)
app.mount("/mcp", mcp_server.app)

# app/tools.py
from pydantic import BaseModel, Field
from http_mcp.types import Arguments
from app.context import AppContext

class SearchInput(BaseModel):
    query: str = Field(description="The search query")

class SearchOutput(BaseModel):
    results: list[str] = Field(description="Search result titles")

async def search(args: Arguments[SearchInput]) -> SearchOutput:
    """Search via an external API."""
    ctx = args.get_state_key("ctx", AppContext)
    resp = await ctx.http_client.get("/search", params={"q": args.inputs.query})
    resp.raise_for_status()
    return SearchOutput(results=[r["title"] for r in resp.json()["items"]])

In-Memory-Cache

Nutzen Sie veränderlichen Zustand wie Caches oder Zähler über Tool-Aufrufe innerhalb desselben Server-Lebenszyklus hinweg gemeinsam:

# app/context.py
from dataclasses import dataclass, field

@dataclass
class AppContext:
    cache: dict[str, str] = field(default_factory=dict)
    request_count: int = 0

# app/tools.py
from pydantic import BaseModel, Field
from http_mcp.types import Arguments
from app.context import AppContext

class LookupInput(BaseModel):
    key: str = Field(description="The cache key to look up")

class LookupOutput(BaseModel):
    value: str | None = Field(description="The cached value, or null if not found")
    total_requests: int = Field(description="Total requests served")

async def lookup(args: Arguments[LookupInput]) -> LookupOutput:
    """Look up a value in the cache."""
    ctx = args.get_state_key("ctx", AppContext)
    ctx.request_count += 1
    return LookupOutput(
        value=ctx.cache.get(args.inputs.key),
        total_requests=ctx.request_count,
    )

Alle Tools, die dieselbe AppContext-Instanz gemeinsam nutzen, sehen die Schreibvorgänge der jeweils anderen sofort, da der Lifespan ein einzelnes gemeinsames Objekt bereitstellt.

Hinweis: Einfache dict- und int-Objekte sind nicht threadsicher. Wenn Ihre Tools parallel ausgeführt werden (z. B. synchrone Tools, die über Threads verteilt werden), schützen Sie gemeinsamen veränderlichen Zustand mit einem asyncio.Lock oder verwenden Sie threadsichere Datenstrukturen.

Request-Zugriff

Sie können aus Ihren Tools auf das eingehende Request-Objekt zugreifen. Das Request-Objekt wird an jeden Tool-Aufruf übergeben und kann verwendet werden, um auf Header, Cookies und andere Request-Daten zuzugreifen (z. B. request.state, request.scope).

from pydantic import BaseModel, Field
from http_mcp.types import Arguments

class MyToolArguments(BaseModel):
    question: str = Field(description="The question to answer")

class MyToolOutput(BaseModel):
    answer: str = Field(description="The answer to the question")


async def my_tool(args: Arguments[MyToolArguments]) -> MyToolOutput:
    # Access the request
    auth_header = args.request.headers.get("Authorization")
    ...

    return MyToolOutput(answer=f"Hello, {args.inputs.question}!")

# Use MCPServer:
from http_mcp.server import MCPServer

mcp_server = MCPServer(
    name="my-server",
    version="1.0.0",
    tools=(my_tool,),
)

Prompts

Sie können interaktive Vorlagen hinzufügen, die auf Benutzerwunsch aufgerufen werden. Prompts unterstützen jetzt den Zugriff auf den Lifespan-Zustand, ähnlich wie Tools.

Einfaches Prompt-Beispiel

1. Definieren Sie die Argumente für die Prompts:

from pydantic import BaseModel, Field

from http_mcp.types import Arguments, Prompt, PromptMessage, TextContent


class GetAdvice(BaseModel):
    topic: str = Field(description="The topic to get advice on")
    include_actionable_steps: bool = Field(
        description="Whether to include actionable steps in the advice", default=False
    )


def get_advice(args: Arguments[GetAdvice]) -> tuple[PromptMessage, ...]:
    """Get advice on a topic."""
    template = """
    You are a helpful assistant that can give advice on {topic}.
    """
    if args.inputs.include_actionable_steps:
        template += """
        The advice should include actionable steps.
        """
    return (
        PromptMessage(
            role="user",
            content=TextContent(
                text=template.format(topic=args.inputs.topic)
            ),
        ),
    )


PROMPTS = (
    Prompt(
        func=get_advice,
        arguments_type=GetAdvice,
    ),
)

2. Instanziieren Sie den Server:

from starlette.applications import Starlette

from app.prompts import PROMPTS
from http_mcp.server import MCPServer

app = Starlette()
mcp_server = MCPServer(tools=(), prompts=PROMPTS, name="test", version="1.0.0")

app.mount(
    "/mcp",
    mcp_server.app,
)

Prompts ohne Argumente

Sie können Prompts definieren, die keine Eingabeargumente erfordern:

from http_mcp.types import Prompt, PromptMessage, TextContent

def help_prompt() -> tuple[PromptMessage, ...]:
    """Use this prompt to get general help."""
    return (
        PromptMessage(
            role="user",
            content=TextContent(
                text="You are a helpful assistant. Help the user with their task."
            ),
        ),
    )

PROMPTS = (
    Prompt(
        func=help_prompt,
        arguments_type=type(None),  # No arguments required
    ),
)

Alternativ können Sie die Klasse NoArguments verwenden:

from http_mcp.types import Arguments, NoArguments, Prompt, PromptMessage, TextContent

def help_prompt_with_context(args: Arguments[NoArguments]) -> tuple[PromptMessage, ...]:
    """Use this prompt to get help with access to context."""
    # You can still access request and state
    context = args.get_state_key("context", Context)
    return (
        PromptMessage(
            role="user",
            content=TextContent(text="You are a helpful assistant."),
        ),
    )

PROMPTS = (
    Prompt(
        func=help_prompt_with_context,
        arguments_type=NoArguments,
    ),
)

Prompts mit Lifespan-Zustand

from pydantic import BaseModel, Field
from http_mcp.types import Arguments, Prompt, PromptMessage, TextContent
from app.context import Context

class GetAdvice(BaseModel):
    topic: str = Field(description="The topic to get advice on")

def get_advice_with_context(args: Arguments[GetAdvice]) -> tuple[PromptMessage, ...]:
    """Get advice on a topic with context awareness."""
    # Access the context from lifespan state
    context = args.get_state_key("context", Context)
    called_tools = context.get_called_tools()
    template = """
    You are a helpful assistant that can give advice on {topic}.
    Previously called tools: {tools}
    """

    return (
        PromptMessage(
            role="user",
            content=TextContent(
                text=template.format(
                    topic=args.inputs.topic,
                    tools=", ".join(called_tools) if called_tools else "none"
                )
            )
        ),
    )

PROMPTS_WITH_CONTEXT = (
    Prompt(
        func=get_advice_with_context,
        arguments_type=GetAdvice,
    ),
)

Prompts mit Autorisierungs-Scopes

Sie können den Zugriff auf Prompts basierend auf Authentifizierungs-Scopes einschränken:

from http_mcp.types import Arguments, NoArguments, Prompt, PromptMessage, TextContent

def private_prompt(args: Arguments[NoArguments]) -> tuple[PromptMessage, ...]:
    """Private prompt that is only accessible to authenticated users."""
    return (
        PromptMessage(
            role="user",
            content=TextContent(text="This is a private prompt."),
        ),
    )

def admin_prompt(args: Arguments[NoArguments]) -> tuple[PromptMessage, ...]:
    """Admin prompt accessible to users with admin or superuser scope."""
    return (
        PromptMessage(
            role="user",
            content=TextContent(text="This is an admin prompt."),
        ),
    )

PROMPTS = (
    Prompt(
        func=private_prompt,
        arguments_type=NoArguments,
        scopes=("private",),  # Only accessible with 'private' scope
    ),
    Prompt(
        func=admin_prompt,
        arguments_type=NoArguments,
        scopes=("admin", "superuser"),  # Accessible with either scope
    ),
)

Hinweis: Sie müssen in Ihrer Starlette-App eine Authentifizierungs-Middleware einrichten, damit Scopes ordnungsgemäß funktionieren.

STDIO-Transport

Zusätzlich zum HTTP-Transport unterstützt der Server den STDIO-Transport für die Kommunikation. Dies ist nützlich für Kommandozeilenanwendungen und Integrationen, die über Standardeingabe/-ausgabe kommunizieren.

Verwendung des STDIO-Transports

import asyncio
import os
from http_mcp.server import MCPServer
from app.tools import TOOLS
from app.prompts import PROMPTS

mcp_server = MCPServer(
    tools=TOOLS,
    prompts=PROMPTS,
    name="test",
    version="1.0.0"
)

# Run the server with STDIO transport
async def main() -> None:
    request_headers = {
        "Authorization": f"Bearer {os.getenv('MCP_TOKEN', '')}",
        "X-Custom-Header": "value",
    }
    await mcp_server.serve_stdio(request_headers)

asyncio.run(main())

Der Parameter request_headers ermöglicht es Ihnen, Header zu übergeben, die in den Request-Kontext aufgenommen werden, wodurch Authentifizierung und andere headerbasierte Funktionen auch bei Verwendung des STDIO-Transports aktiviert werden.

Authentifizierung und Autorisierung

Die Bibliothek integriert sich in das Authentifizierungssystem von Starlette, um eine Scope-basierte Autorisierung für Tools und Prompts bereitzustellen.

Einrichten der Authentifizierungs-Middleware

import contextlib
from collections.abc import AsyncIterator
from typing import TypedDict
from starlette.applications import Starlette
from starlette.authentication import (
    AuthCredentials,
    AuthenticationBackend,
    BaseUser,
    SimpleUser,
)
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.requests import HTTPConnection

from http_mcp.server import MCPServer
from app.context import Context
from app.tools import TOOLS
from app.prompts import PROMPTS


class BasicAuthBackend(AuthenticationBackend):
    def __init__(self, granted_scopes: tuple[str, ...] = ("authenticated",)) -> None:
        self.granted_scopes = granted_scopes
        super().__init__()

    async def authenticate(
        self, conn: HTTPConnection
    ) -> tuple[AuthCredentials, BaseUser] | None:
        # Implement your authentication logic here
        # For example, check Bearer token, API key, etc.
        auth_header = conn.headers.get("Authorization")
        if not auth_header:
            return None

        # Validate token and return credentials with scopes
        return AuthCredentials(self.granted_scopes), SimpleUser("username")


class State(TypedDict):
    context: Context


@contextlib.asynccontextmanager
async def lifespan(_app: Starlette) -> AsyncIterator[State]:
    yield {"context": Context()}


mcp_server = MCPServer(
    tools=TOOLS,
    prompts=PROMPTS,
    name="test",
    version="1.0.0"
)

app = Starlette(
    lifespan=lifespan,
    middleware=[
        Middleware(
            AuthenticationMiddleware,
            backend=BasicAuthBackend(granted_scopes=("private", "admin")),
        ),
    ],
)
app.mount("/mcp", mcp_server.app)

Wie Scopes funktionieren

1. Authentifizierungs-Middleware: Die Middleware authentifiziert jede Anfrage und weist dem Benutzer über AuthCredentials Scopes zu.

2. Tool-/Prompt-Scopes: Beim Definieren von Tools oder Prompts können Sie erforderliche Scopes über den Parameter scopes angeben.

3. Zugriffskontrolle: Der Server filtert Tools und Prompts automatisch anhand der dem Benutzer gewährten Scopes. Tools und Prompts ohne die erforderlichen Scopes sind in Auflistungen nicht sichtbar und können nicht aufgerufen werden.

4. Mehrere Scopes: Wenn Sie mehrere Scopes angeben (z. B. scopes=("admin", "superuser")), benötigt der Benutzer mindestens einen dieser Scopes, um auf das Tool oder den Prompt zuzugreifen.

API-Referenz

Tool-Klasse

Die Klasse Tool wird verwendet, um Tools zu definieren, die von Clients aufgerufen werden können.

Parameter:

• func: Die aufzurufende Funktion. Kann synchron oder asynchron sein. Die Funktion kann entweder:

  • Einen Arguments[TInputs]-Parameter akzeptieren

  • Keine Parameter akzeptieren

• inputs: Die Pydantic-Modellklasse für die Eingabevalidierung. Verwenden Sie type(None) oder NoArguments für Tools ohne Eingaben

• output: Die Pydantic-Modellklasse für die Ausgabevalidierung

• return_error_message (bool): Wenn True, geben Tool-Fehler ErrorMessage zurück, anstatt Ausnahmen auszulösen (Standard: False)

• scopes (tuple[str, ...]): Erforderliche Authentifizierungs-Scopes für den Zugriff auf dieses Tool (Standard: leeres Tupel)

Eigenschaften:

• name: Der Funktionsname (abgeleitet von func.__name__)

• title: Ein menschenlesbarer Titel (abgeleitet vom Funktionsnamen)

• description: Der Docstring der Funktion

• input_schema: JSON-Schema für die Eingabeparameter

• output_schema: JSON-Schema für die Ausgabe

Prompt-Klasse

Die Klasse Prompt wird verwendet, um Prompts zu definieren, die von Clients aufgerufen werden können.

Parameter:

• func: Die aufzurufende Funktion. Kann synchron oder asynchron sein. Die Funktion kann entweder:

  • Einen Arguments[TArguments]-Parameter akzeptieren

  • Keine Parameter akzeptieren

  • Muss tuple[PromptMessage, ...] zurückgeben

• arguments_type: Die Pydantic-Modellklasse für die Argumentvalidierung. Verwenden Sie type(None) oder NoArguments für Prompts ohne Argumente

• scopes (tuple[str, ...]): Erforderliche Authentifizierungs-Scopes für den Zugriff auf diesen Prompt (Standard: leeres Tupel)

Eigenschaften:

• name: Der Funktionsname (abgeleitet von func.__name__)

• title: Ein menschenlesbarer Titel (abgeleitet vom Funktionsnamen)

• description: Der Docstring der Funktion

• arguments: Tupel von PromptArgument-Objekten, die die Argumente des Prompts definieren

Arguments-Klasse

Die Klasse Arguments wird an Tool- und Prompt-Funktionen übergeben, um Zugriff auf Eingaben, Request und Zustand zu ermöglichen.

Parameter:

• request: Das Starlette-Request-Objekt

• inputs: Die validierten Eingabe-/Argumentdaten (Typ hängt von der Tool-/Prompt-Definition ab)

Methoden:

• get_state_key(key: str, _object_type: type[TKey]) -> TKey: Greift auf einen Wert aus dem Lifespan-Zustand zu. Löst ServerError aus, wenn der Schlüssel nicht existiert.

NoArguments-Klasse

Ein leeres Pydantic-Modell, das als klarere Alternative zu type(None) verwendet werden kann, wenn Sie Tools oder Prompts ohne Argumente definieren.

from http_mcp.types import NoArguments

# Use this instead of type(None)
Tool(func=my_func, inputs=NoArguments, output=MyOutput)

OAuth-2.1-Autorisierung (auth_mcp)

Das Paket auth_mcp fügt Ihrem MCP-Server standardskonforme OAuth-2.1-Autorisierung hinzu. Installieren Sie es mit dem auth-Extra:

pip install http-mcp[auth]

Schnellstart

from http_mcp.server import MCPServer
from auth_mcp.resource_server import (
    ProtectedMCPAppConfig,
    TokenInfo,
    TokenValidator,
    create_protected_mcp_app,
)
from auth_mcp.types import ProtectedResourceMetadata


class MyTokenValidator(TokenValidator):
    async def validate_token(
        self, token: str, resource: str | None = None
    ) -> TokenInfo | None:
        # Validate against your authorization server
        ...


mcp_server = MCPServer(name="my-server", version="1.0.0", tools=MY_TOOLS)

config = ProtectedMCPAppConfig(
    mcp_server=mcp_server,
    token_validator=MyTokenValidator(),
    resource_endpoint=ProtectedResourceMetadata(
        resource="https://mcp.example.com",
        authorization_servers=("https://auth.example.com",),
    ),
)

app = create_protected_mcp_app(config)

Das bietet Ihnen:

• Bearer-Token-Validierung auf allen MCP-Endpunkten (standardmäßig sicher)

• /.well-known/oauth-protected-resource-Discovery-Endpunkt (RFC 9728)

• WWW-Authenticate-Header bei 401/403 mit resource_metadata-Parameter

• Sicherheits-Header (HSTS, nosniff, no-store)

• Optionale benutzerdefinierte Middleware über den Parameter middlewares

Die vollständige Dokumentation, Best Practices und Details zur Sicherheitsfläche finden Sie unter auth_mcp README.

Sicherheitsflächen nach Endpunkt

POST /mcp — MCP-JSON-RPC-Endpunkt

• Authentifizierung — Bei Verwendung von auth_mcp werden Bearer-Tokens aus dem Authorization-Header extrahiert und über TokenValidator validiert. Tokens, die 2048 Zeichen überschreiten oder Zeichen außerhalb des RFC-6750-b64token-Musters enthalten, werden abgelehnt, bevor sie den Validator erreichen. Ohne auth_mcp wird die Authentifizierung von der AuthenticationMiddleware von Starlette übernommen.

• Autorisierung — Scope-basierte Filterung über has_required_scope() von Starlette. Tools und Prompts ohne passende Scopes werden in Auflistungen ausgeblendet und beim Aufruf blockiert. Die Request-Header-Validierung löst Tool-Schemas über dieselbe Scope-Prüfung auf, sodass ein Aufrufer, dem ein Tool verborgen ist, dessen x-mcp-header-Argumente auch nicht aus einer Mismatch-Meldung erfahren kann.

• Eingabevalidierung — JSON-RPC-Nachrichten werden von Pydantic validiert. Der Request-Body ist auf 4 MB begrenzt, was beim Lesen durchgesetzt wird: Ein zu großer Content-Length-Wert wird abgelehnt, bevor der Body überhaupt gelesen wird, und ein Body, der die Grenze mitten im Stream überschreitet, wird ab diesem Punkt nicht mehr gepuffert. Content-Type wird streng geprüft (nur application/json, Medientyp-Parameter werden ignoriert).

• Fehlerbehandlung — Tool- und Prompt-Namen werden in Fehlermeldungen auf 100 Zeichen gekürzt. Pydantic-Validierungsfehler werden bereinigt, bevor sie in Antworten aufgenommen werden.

• Antwort-Header — X-Content-Type-Options: nosniff, Cache-Control: no-store auf allen Antworten. auth_mcp fügt zusätzlich Strict-Transport-Security: max-age=31536000; includeSubDomains hinzu.

GET /.well-known/oauth-protected-resource — Discovery-Endpunkt (auth_mcp)

• Authentifizierung — Unterliegt derselben Auth-Middleware wie /mcp. Wenn require_authentication=True (Standard), ist ein gültiges Token erforderlich. Setzen Sie den Wert auf False, wenn Clients den Autorisierungsserver entdecken müssen, bevor sie sich authentifizieren.

• Eingabevalidierung — Nur GET ist erlaubt; andere Methoden geben 405 Method Not Allowed zurück.

• Ausgabe — Wird beim Start einmalig aus einem eingefrorenen ProtectedResourceMetadata-Modell serialisiert. URI-Felder werden über AnyHttpUrl von Pydantic als HTTP/HTTPS-URLs validiert.

WWW-Authenticate-Antwortheader (auth_mcp)

• Header-Injection — Alle Parameterwerte (realm, resource_metadata, scope, error, error_description) werden bereinigt: CR/LF-Zeichen werden entfernt, Backslash und doppelte Anführungszeichen werden gemäß den RFC-7230-Quoted-String-Regeln maskiert.

• Informationspreisgabe — Fehlerantworten verwenden generische Meldungen ("Authentication required"). Die Details der ursprünglichen AuthenticationError werden verworfen. Fehlercodes (invalid_token bei 401) folgen RFC 6750, ohne internen Zustand preiszugeben.

STDIO-Transport

• Nachrichtengröße — Auf 4 MB begrenzt, entsprechend dem HTTP-Transport.

• Protokollierung — Nachrichten werden in Debug-Logs auf 500 Zeichen gekürzt, um eine Überflutung der Logs zu verhindern. Token-Werte werden nie protokolliert.

• Header — Request-Header werden in das korrekte ASGI-Format list[tuple[bytes, bytes]] konvertiert.

Installation

Erfordert Python 3.12+ (verwendet die Typ-Parameter-Syntax von PEP 695).

Installieren Sie das Paket mit pip oder uv:

pip install http-mcp

Mit Unterstützung für OAuth-2.1-Autorisierung:

pip install http-mcp[auth]

oder

uv add http-mcp

Lizenz

Dieses Projekt ist unter der MIT-Lizenz lizenziert. Einzelheiten finden Sie in der LICENSE-Datei.