de en es zh

Simple HTTP MCP Server

Name: Simple HTTP MCP Server
Author: yeison-liscano

by yeison-liscano

Overview Schema Related Servers Score Discussions

Python

Remote

Implementierung eines einfachen HTTP-MCP-Servers

Dieses Projekt bietet eine leichtgewichtige Server-Implementierung für das Model Context Protocol (MCP) über HTTP. Es ermöglicht Ihnen, 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 vorgesehen (siehe Demo).

Inhaltsverzeichnis

Related MCP server: Python REPL MCP Server

Funktionen

MCP-Protokoll-konform: Implementiert die MCP-Spezifikation für die Entdeckung und Ausführung von Tools und Prompts. Keine Unterstützung für Benachrichtigungen.
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 Anfragenverarbeitung.
Typsicher: Nutzt Pydantic für robuste Datenvalidierung und Serialisierung.
Server-Zustandsverwaltung: Zugriff auf den geteilten Zustand über den Lebensdauer-Kontext (Lifespan) unter Verwendung der Methode get_state_key.
Anfrage-Zugriff: Zugriff auf das eingehende Anfrageobjekt aus Ihren Tools und Prompts.
Autorisierungs-Scopes: Unterstützung für Scope-basierte Autorisierung unter Verwendung des Authentifizierungssystems von Starlette.
Fehlerbehandlung: Tools können optional Fehlermeldungen zurückgeben, anstatt Exceptions 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].

Server-Architektur

Die Bibliothek bietet eine einzelne MCPServer-Klasse, die die Lebensdauer (Lifespan) nutzt, um den geteilten Zustand über den gesamten Anwendungslebenszyklus zu verwalten.

MCPServer

Der MCPServer ist für die Zusammenarbeit mit dem Lebensdauer-System von Starlette zur Verwaltung des geteilten Serverzustands konzipiert.

Hauptmerkmale:

Lebensdauer-basiert: Verwendet die Lebensdauer-Ereignisse von Starlette, um den geteilten Serverzustand zu initialisieren und zu verwalten.
Zustand auf Anwendungsebene: Der Zustand bleibt über den gesamten Anwendungslebenszyklus bestehen, nicht pro Anfrage.
Flexibel: Kann mit jeder benutzerdefinierten Kontextklasse verwendet werden, die im Lebensdauer-Zustand gespeichert ist.

Konstruktor-Parameter:

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

Beispielanwendung:

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)

Tools

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

Einfaches Tool-Beispiel

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.

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"]

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 Klasse NoArguments für eine bessere Übersichtlichkeit 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 Exceptions 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 einen ToolInvocationError auszulösen.

Tools mit Autorisierungs-Scopes

Sie können den Tool-Zugriff basierend auf Authentifizierungs-Scopes 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 eine Authentifizierungs-Middleware in Ihrer Starlette-App einrichten, damit Scopes ordnungsgemäß funktionieren. Das Feld scopes in Tool ist das primäre Autorisierungstor — das Framework filtert Tools vor dem Aufruf nach Scopes. Die Aufrufe von raise ToolInvocationError(...) innerhalb der Tool-Funktionen oben sind optionale Sicherheitsprüfungen, die eine ordnungsgemäße Fehlerantwort an den Client zurückgeben, anstatt stillschweigend zu scheitern.

Server-Zustandsverwaltung

Der Server verwendet das Lebensdauer-System von Starlette, um den geteilten Zustand über den gesamten Anwendungslebenszyklus zu verwalten. Der Zustand wird beim Start der Anwendung initialisiert und bleibt bis zum Herunterfahren bestehen. Der Zugriff auf den Kontext erfolgt über die Methode get_state_key des Arguments-Objekts.

Dies ist nützlich, um Ressourcen wie Datenbank-Verbindungspools, HTTP-Clients, Caches oder jeden anderen Anwendungszustand über Tools hinweg zu teilen.

Datenbank-Verbindungspool

Das gängigste Muster: Initialisieren Sie einen Verbindungspool beim Start, teilen Sie ihn über alle Tools hinweg und schließen Sie ihn beim Herunterfahren:

# 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"])

Geteilter HTTP-Client

Teilen Sie einen einzelnen httpx.AsyncClient über Tools hinweg, 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

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

# 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 teilen, sehen die Schreibvorgänge der anderen sofort, da die Lebensdauer ein einzelnes geteiltes Objekt liefert.

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

Anfrage-Zugriff

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

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 durch Benutzerauswahl aufgerufen werden. Prompts unterstützen jetzt den Zugriff auf den Lebensdauer-Zustand, ähnlich wie Tools.

Einfaches Prompt-Beispiel

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

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 Lebensdauer-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 Prompt-Zugriff 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 eine Authentifizierungs-Middleware in Ihrer Starlette-App 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 Befehlszeilenanwendungen und Integrationen, die über die Standard-Ein-/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 Anfragekontext aufgenommen werden, wodurch Authentifizierung und andere Header-basierte Funktionen auch bei Verwendung des STDIO-Transports ermöglicht werden.

Authentifizierung und Autorisierung

Die Bibliothek lässt sich in das Authentifizierungssystem von Starlette integrieren, 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)

Funktionsweise der Scopes

Authentifizierungs-Middleware: Die Middleware authentifiziert jede Anfrage und weist dem Benutzer über AuthCredentials Scopes zu.
Tool-/Prompt-Scopes: Beim Definieren von Tools oder Prompts können Sie erforderliche Scopes über den Parameter scopes angeben.
Zugriffskontrolle: Der Server filtert Tools und Prompts automatisch basierend auf den gewährten Scopes des Benutzers. Tools und Prompts ohne die erforderlichen Scopes sind in Auflistungen nicht sichtbar und können nicht aufgerufen werden.
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 Tool-Klasse 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 Exceptions 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 Prompt-Klasse 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 Arguments-Klasse wird an Tool- und Prompt-Funktionen übergeben, um Zugriff auf Eingaben, Anfrage und Zustand zu gewähren.

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: Zugriff auf einen Wert aus dem Lebensdauer-Zustand. 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 Tools oder Prompts ohne Argumente definiert werden.

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 auth_mcp-Paket fügt Ihrem MCP-Server eine standardkonforme 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)

Dies bietet Ihnen:

Bearer-Token-Validierung auf allen MCP-Endpunkten (standardmäßig sicher)
/.well-known/oauth-protected-resource-Entdeckungs-Endpunkt (RFC 9728)
WWW-Authenticate-Header bei 401/403 mit resource_metadata-Parameter
Sicherheits-Header (HSTS, nosniff, no-store)
Optionale benutzerdefinierte Middleware über den middlewares-Parameter

Für die vollständige Dokumentation, Best Practices und Details zur Sicherheitsoberfläche siehe auth_mcp README.

Sicherheitsoberflächen nach Endpunkt

`POST /mcp` — MCP JSON-RPC-Endpunkt

Authentifizierung — Bei Verwendung von auth_mcp werden Bearer-Token aus dem Authorization-Header extrahiert und über TokenValidator validiert. Token, 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 durch die AuthenticationMiddleware von Starlette gehandhabt.
Autorisierung — Scope-

Install Server

security – no known vulnerabilities

license - permissive license

quality - not tested

How are these scores calculated?

Resources

GitHub Repository

Need Help?

Reddit Discussion

Related Servers

Unclaimed servers have limited discoverability.

Looking for Admin?

If you are the server author, to access and configure the admin panel.

Tools

Appeared in Searches

Simple MCP example using HTTP transport

Latest Blog Posts

Tool Definition Quality Score (TDQS)
By punkpeye on April 3, 2026.
mcp
The Hackers Who Tracked My Sleep Cycle
By punkpeye on March 26, 2026.
security
Open Source Has a Bot Problem
By punkpeye on March 19, 2026.
open source

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/yeison-liscano/http_mcp'

If you have feedback or need assistance with the MCP directory API, please join our Discord server

Implementierung eines einfachen HTTP-MCP-Servers

Inhaltsverzeichnis

Funktionen

Server-Architektur

MCPServer

Tools

Einfaches Tool-Beispiel

Tools ohne Argumente

Tools mit Fehlerbehandlung

Tools mit Autorisierungs-Scopes

Server-Zustandsverwaltung

Datenbank-Verbindungspool

Geteilter HTTP-Client

In-Memory-Cache

Anfrage-Zugriff

Prompts

Einfaches Prompt-Beispiel

Prompts ohne Argumente

Prompts mit Lebensdauer-Zustand

Prompts mit Autorisierungs-Scopes

STDIO-Transport

Verwendung des STDIO-Transports

Authentifizierung und Autorisierung

Einrichten der Authentifizierungs-Middleware

Funktionsweise der Scopes

API-Referenz

Tool-Klasse

Prompt-Klasse

Arguments-Klasse

NoArguments-Klasse

OAuth 2.1 Autorisierung (auth_mcp)

Schnellstart

Sicherheitsoberflächen nach Endpunkt

POST /mcp — MCP JSON-RPC-Endpunkt

Resources

Looking for Admin?

Tools

Appeared in Searches

Latest Blog Posts

MCP directory API

`POST /mcp` — MCP JSON-RPC-Endpunkt