Implementación de un servidor MCP HTTP simple

Este proyecto proporciona una implementación ligera de servidor para el Model Context Protocol (MCP) sobre HTTP. Permite exponer funciones de Python como herramientas y prompts que pueden descubrirse y ejecutarse de forma remota a través de una interfaz JSON-RPC. Está pensado para usarse con una aplicación Starlette o FastAPI (consulte la demo).

Tabla de contenidos

• Características

• Instalación

• Arquitectura del servidor

• Versión del protocolo

  • La forma de las solicitudes de 2026-07-28

  • Sugerencias de caché

  • Validación de origen

  • Reflejo de parámetros de herramientas en cabeceras

• Herramientas

  • Ejemplo básico de herramienta

  • Herramientas sin argumentos

  • Herramientas con gestión de errores

  • Herramientas con ámbitos de autorización

• Gestión del estado del servidor

• Acceso a la solicitud

• Prompts

  • Ejemplo básico de prompt

  • Prompts sin argumentos

  • Prompts con estado de ciclo de vida

  • Prompts con ámbitos de autorización

• Transporte STDIO

• Autenticación y autorización

• Autorización OAuth 2.1 (auth_mcp)

• Referencia de la API

• Superficies de seguridad por endpoint

• Licencia

Related MCP server: remote-mcp

Características

• Conforme con el protocolo MCP: Implementa la especificación MCP para el descubrimiento y la ejecución de herramientas y prompts. No admite notificaciones.

• Una única revisión del protocolo: Solo habla la revisión sin estado 2026-07-28 — server/discover, _meta por solicitud, sin handshake, sin sesión. Una única ruta de despacho significa que una solicitud no puede seleccionar un manejo más débil declarando una revisión más antigua.

• Transporte HTTP y STDIO: Utiliza HTTP (solicitudes POST) o STDIO para la comunicación.

• Soporte asíncrono: Construido sobre Starlette o FastAPI para el manejo asíncrono de solicitudes.

• Seguridad de tipos: Aprovecha Pydantic para una validación y serialización robusta de datos.

• Gestión del estado del servidor: Acceda al estado compartido a través del contexto de ciclo de vida usando el método get_state_key.

• Acceso a la solicitud: Acceda al objeto de solicitud entrante desde sus herramientas y prompts.

• Ámbitos de autorización: Soporte para autorización basada en ámbitos usando el sistema de autenticación de Starlette.

• Gestión de errores: Las herramientas pueden devolver opcionalmente mensajes de error en lugar de lanzar excepciones.

• Autorización OAuth 2.1: Paquete opcional auth_mcp con validación de tokens Bearer, metadatos de recursos protegidos (RFC 9728) y respuestas de error WWW-Authenticate. Instale con pip install http-mcp[auth].

Arquitectura del servidor

La biblioteca proporciona una única clase MCPServer que utiliza el ciclo de vida para gestionar el estado compartido durante todo el ciclo de vida de la aplicación.

MCPServer

El MCPServer está diseñado para funcionar con el sistema de ciclo de vida de Starlette para gestionar el estado compartido del servidor.

Características clave:

• Basado en ciclo de vida: Utiliza los eventos de ciclo de vida de Starlette para inicializar y gestionar el estado compartido del servidor

• Estado a nivel de aplicación: El estado persiste durante todo el ciclo de vida de la aplicación, no por solicitud

• Flexible: Puede usarse con cualquier clase de contexto personalizada almacenada en el estado del ciclo de vida

Parámetros del constructor:

• name (str): El nombre de su servidor MCP

• version (str): La versión de su servidor MCP

• tools (tuple[Tool, ...]): Tupla de herramientas a exponer (por defecto: tupla vacía)

• prompts (tuple[Prompt, ...]): Tupla de prompts a exponer (por defecto: tupla vacía)

• instructions (str | None): Instrucciones opcionales para asistentes de IA sobre cómo usar este servidor

• cache_ttl_ms (int): Sugerencia de frescura en milisegundos enviada con los resultados de tools/list, prompts/list y server/discover (por defecto: 300000). Use 0 para indicar a los clientes que nunca almacenen en caché. Consulte Sugerencias de caché.

• cache_scope ("public" | "private" | None): Si las cachés compartidas pueden reutilizar esos resultados entre contextos de autorización. Se deriva automáticamente cuando se omite. Consulte Sugerencias de caché.

• allowed_origins (tuple[str, ...]): Orígenes que acepta el transporte HTTP (por defecto: vacío, lo que significa que la comprobación está deshabilitada). Consulte Validación de origen.

• require_origin (bool): Si una solicitud que no lleva ninguna cabecera Origin se rechaza cuando allowed_origins está configurado (por defecto: False). Consulte Validación de origen.

Ejemplo de uso:

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)

Versión del protocolo

El servidor implementa exactamente una revisión del protocolo, 2026-07-28, y cada solicitud recorre la misma ruta. No hay negociación de versión ni un segundo conjunto de reglas al que una solicitud pueda acogerse.

Cambio importante en 0.17.0. Se eliminó el soporte para las revisiones basadas en sesión 2025-11-25, 2025-06-18 y 2025-03-26, junto con initialize, notifications/initialized y ping. Un cliente que solo hable esas revisiones ya no puede comunicarse con este servidor. Servir una única revisión es también lo que hace fiables las cabeceras de metadatos de solicitud que se indican a continuación: mientras coexistieron dos eras, una solicitud podía omitir las comprobaciones de cabecera declarando la más antigua, por lo que un intermediario que enrutara según Mcp-Method podía desincronizarse del servidor que actuaba sobre el cuerpo.

Cambio importante en 0.18.0. ServerInterface.get_tool_input_schema ahora recibe la Request, por lo que los ámbitos de autorización se respetan antes del despacho; las implementaciones de la interfaz deben actualizarse, mientras que los usuarios de MCPServer no se ven afectados. Los valores reflejados de Mcp-Param-* se comparan textualmente en lugar de numéricamente, por lo que una cabecera que indique 3.0 para "replicas": 3 ahora recibe -32020. Cada método notifications/* devuelve 404, ya que la revisión no define ninguno.

La forma de las solicitudes de 2026-07-28

La revisión no tiene concepto de sesión. En la práctica:

• Sin handshake. Cada solicitud reafirma su versión del protocolo y las capacidades del cliente en _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 y clientCapabilities son obligatorios; omitir cualquiera de ellos produce un -32602 y HTTP 400. Cualquier otra versión produce un -32022 cuyo data.supported enumera la única revisión que habla este servidor.

• server/discover reemplaza a initialize para el descubrimiento de capacidades. Informa de la versión admitida, las capacidades, las instrucciones y la identidad del servidor en una sola llamada, y se responde sin ninguna solicitud previa:

  {
    "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"
  }

• Cada resultado lleva resultType: "complete" y un bloque _meta que nombra al servidor.

• initialize, notifications/initialized, ping y logging/setLevel no existen, junto con la maquinaria de sesión y reanudación SSE. Devuelven -32601 con HTTP 404. Las notificaciones JSON-RPC — un mensaje notifications/* sin id — siguen recibiendo 202 Accepted y sin cuerpo, porque JSON-RPC prohíbe responder a ellas.

• Cabeceras de solicitud obligatorias. Cada POST debe enviar MCP-Protocol-Version y Mcp-Method, además de Mcp-Name en tools/call y prompts/get. Cada una debe coincidir con el valor correspondiente del cuerpo, o la solicitud se rechaza con -32020 (HeaderMismatch) y HTTP 400 — esto impide que un proxy enrute según un valor mientras el servidor actúa sobre otro. Los valores que no pueden expresarse como ASCII simple usan el sobre =?base64?...?=, que el servidor decodifica antes de comparar.

• Las herramientas y prompts desconocidos informan de -32602, que es lo que prescriben las especificaciones de herramientas y prompts. -32002 fue retirado por esta revisión.

• Mcp-Session-Id y Last-Event-ID se ignoran, y GET/DELETE en el endpoint MCP devuelven 405 Method Not Allowed.

Las solicitudes de múltiples idas y vueltas (elicitación, muestreo, raíces) y subscriptions/listen no están implementadas: este servidor no expone características dependientes de la entrada del cliente y declara listChanged: false, por lo que ninguna de ellas le aplica.

Sugerencias de caché

Los resultados de tools/list, prompts/list y server/discover llevan ttlMs y cacheScope para que los clientes puedan evitar volver a obtener una lista que no ha cambiado:

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
)

Las herramientas y los prompts son fijos cuando se construye MCPServer, por lo que ttlMs realmente limita cuánto tiempo puede un cliente no detectar un redespliegue, más que cuánto tiempo son estables los datos. Establézcalo en 0 para pedir a los clientes que nunca almacenen en caché.

cache_scope se deriva cuando se omite: "private" si alguna herramienta o prompt está restringido por ámbito — la lista varía entonces según el llamante, por lo que una caché compartida no debe reutilizarla entre contextos de autorización — y "public" en caso contrario. Anúlelo si su despliegue sabe más. Tenga en cuenta que cacheScope solo rige el almacenamiento en caché; nunca sustituye a las comprobaciones de ámbito por herramienta.

Validación de origen

Los navegadores adjuntan una cabecera Origin, que es lo que permite a un servidor rechazar solicitudes introducidas mediante DNS rebinding. La comprobación está desactivada por defecto para que los despliegues existentes sigan funcionando; actívela dondequiera que el endpoint sea accesible desde un navegador:

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

Una solicitud cuyo Origin esté presente y no esté en la lista recibe 403 Forbidden. Las solicitudes sin Origin — clientes ordinarios que no son navegadores — no se ven afectadas por defecto, porque los navegadores siempre envían Origin en un POST y el modelo de amenaza del rebinding no cubre a clientes que no son navegadores.

Si el endpoint solo debe atender tráfico de navegador, añada require_origin para rechazar también una solicitud que omita la cabecera, lo que convierte la lista de permitidos en obligatoria en lugar de consultiva:

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

require_origin no hace nada por sí solo — solo refuerza una lista de permitidos que ya está configurada. Cuando ejecute localmente, enlace también a 127.0.0.1 en lugar de 0.0.0.0.

Reflejo de parámetros de herramientas en cabeceras

Una herramienta puede pedir a los clientes que copien valores de argumentos específicos en cabeceras Mcp-Param-*, para que los proxies puedan enrutar o limitar la velocidad según ellos sin analizar el cuerpo. Anote el campo con 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")

Un cliente conforme envía entonces Mcp-Param-Region: us-west1 junto con la llamada, y el servidor lo verifica contra el cuerpo — rechazando la solicitud con -32020 si la cabecera falta, contradice el argumento o se envía cuando el argumento está ausente. Las cabeceras Mcp-Param-* que ninguna anotación reclama se ignoran, ya que se espera que los intermediarios reenvíen las no reconocidas sin tocarlas.

La comparación es textual, contra el valor tal como JSON lo escribe: para "replicas": 3 la cabecera debe indicar exactamente 3, no 3.0, +3 o 3. La coerción numérica los consideraría iguales mientras que un intermediario que enrutara según la cadena de cabecera cruda vería otra cosa, que es la desincronización que el reflejo existe para prevenir.

Solo los campos string, integer y boolean alcanzables a través de una cadena simple de propiedades de objeto pueden anotarse, y no puede haber dos campos que reclamen el mismo nombre de cabecera — una colisión se rechaza cuando se construye el servidor, porque mantener una de las dos anotaciones dejaría la otra silenciosamente sin aplicar. No anote valores sensibles: el contenido de las cabeceras es visible para cada intermediario en la ruta.

Herramientas

Las herramientas son las funciones que el cliente puede llamar.

Ejemplo básico de herramienta

1. Defina los argumentos y la salida de las herramientas:

# 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. Defina las herramientas:

# 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. Instancie el servidor:

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

Herramientas sin argumentos

Puede definir herramientas que no requieran argumentos de entrada:

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

Alternativamente, puede usar la clase NoArguments para mayor claridad:

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

Herramientas con gestión de errores

Las herramientas pueden devolver opcionalmente mensajes de error en lugar de lanzar excepciones:

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

Cuando return_error_message=True, la herramienta devolverá un modelo ErrorMessage con los detalles del error en lugar de lanzar un ToolInvocationError.

Herramientas con ámbitos de autorización

Puede restringir el acceso a las herramientas según los ámbitos de autenticación:

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

Nota: Debes configurar el middleware de autenticación en tu aplicación Starlette para que los scopes funcionen correctamente. El campo scopes en Tool es la puerta de autorización principal: el framework filtra las herramientas por scope antes de la invocación. Las llamadas a raise ToolInvocationError(...) dentro de las funciones de herramienta anteriores son comprobaciones opcionales de defensa en profundidad que devuelven una respuesta de error adecuada al cliente en lugar de fallar silenciosamente.

Gestión del estado del servidor

El servidor utiliza el sistema de ciclo de vida de Starlette para gestionar el estado compartido a lo largo de todo el ciclo de vida de la aplicación. El estado se inicializa cuando la aplicación se inicia y persiste hasta que se apaga. El contexto se accede a través del método get_state_key en el objeto Arguments.

Esto es útil para compartir recursos como pools de conexiones de base de datos, clientes HTTP, cachés o cualquier estado de la aplicación entre herramientas.

Pool de conexiones de base de datos

El patrón más común: inicializar un pool de conexiones al inicio, compartirlo entre todas las herramientas y cerrarlo al apagar:

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

Cliente HTTP compartido

Comparte un único httpx.AsyncClient entre herramientas para reutilizar conexiones y configurar URLs base, cabeceras o tiempos de espera una sola vez:

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

Caché en memoria

Comparte estado mutable como cachés o contadores entre invocaciones de herramientas dentro del mismo ciclo de vida del servidor:

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

Todas las herramientas que comparten la misma instancia de AppContext ven las escrituras de las demás inmediatamente, ya que el ciclo de vida produce un único objeto compartido.

Nota: dict e int simples no son seguros para hilos. Si tus herramientas se ejecutan de forma concurrente (por ejemplo, herramientas síncronas despachadas mediante hilos), protege el estado mutable compartido con un asyncio.Lock o utiliza estructuras de datos seguras para hilos.

Acceso a la solicitud

Puedes acceder al objeto de solicitud entrante desde tus herramientas. El objeto de solicitud se pasa a cada llamada de herramienta y se puede utilizar para acceder a cabeceras, cookies y otros datos de la solicitud (por ejemplo, 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

Puedes añadir plantillas interactivas que se invocan por elección del usuario. Los prompts ahora admiten acceso al estado del ciclo de vida, similar a las herramientas.

Ejemplo básico de prompt

1. Define los argumentos para los 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. Instancia el servidor:

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 sin argumentos

Puedes definir prompts que no requieran argumentos de entrada:

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

Alternativamente, puedes usar la clase NoArguments:

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 con estado del ciclo de vida

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 con scopes de autorización

Puedes restringir el acceso a los prompts según los scopes de autenticación:

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

Nota: Debes configurar el middleware de autenticación en tu aplicación Starlette para que los scopes funcionen correctamente.

Transporte STDIO

Además del transporte HTTP, el servidor admite el transporte STDIO para la comunicación. Esto es útil para aplicaciones de línea de comandos e integraciones que se comunican a través de la entrada/salida estándar.

Uso del transporte STDIO

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

El parámetro request_headers te permite pasar cabeceras que se incluirán en el contexto de la solicitud, lo que permite la autenticación y otras funciones basadas en cabeceras incluso cuando se utiliza el transporte STDIO.

Autenticación y autorización

La biblioteca se integra con el sistema de autenticación de Starlette para proporcionar autorización basada en scopes para herramientas y prompts.

Configuración del middleware de autenticación

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)

Cómo funcionan los scopes

1. Middleware de autenticación: El middleware autentica cada solicitud y asigna scopes al usuario a través de AuthCredentials.

2. Scopes de herramientas/prompts: Al definir herramientas o prompts, puedes especificar los scopes requeridos usando el parámetro scopes.

3. Control de acceso: El servidor filtra automáticamente las herramientas y los prompts según los scopes concedidos al usuario. Las herramientas y los prompts sin los scopes requeridos no son visibles en los listados y no se pueden invocar.

4. Múltiples scopes: Si especificas múltiples scopes (por ejemplo, scopes=("admin", "superuser")), el usuario necesita al menos uno de esos scopes para acceder a la herramienta o al prompt.

Referencia de la API

Clase Tool

La clase Tool se utiliza para definir herramientas que pueden ser invocadas por los clientes.

Parámetros:

• func: La función a invocar. Puede ser síncrona o asíncrona. La función puede:

  • Aceptar un parámetro Arguments[TInputs]

  • No aceptar parámetros

• inputs: La clase de modelo Pydantic para la validación de entrada. Usa type(None) o NoArguments para herramientas sin entradas

• output: La clase de modelo Pydantic para la validación de salida

• return_error_message (bool): Si es True, los errores de la herramienta devuelven ErrorMessage en lugar de lanzar excepciones (por defecto: False)

• scopes (tuple[str, ...]): Scopes de autenticación requeridos para acceder a esta herramienta (por defecto: tupla vacía)

Propiedades:

• name: El nombre de la función (derivado de func.__name__)

• title: Un título legible por humanos (derivado del nombre de la función)

• description: El docstring de la función

• input_schema: Esquema JSON para los parámetros de entrada

• output_schema: Esquema JSON para la salida

Clase Prompt

La clase Prompt se utiliza para definir prompts que pueden ser invocados por los clientes.

Parámetros:

• func: La función a invocar. Puede ser síncrona o asíncrona. La función puede:

  • Aceptar un parámetro Arguments[TArguments]

  • No aceptar parámetros

  • Debe devolver tuple[PromptMessage, ...]

• arguments_type: La clase de modelo Pydantic para la validación de argumentos. Usa type(None) o NoArguments para prompts sin argumentos

• scopes (tuple[str, ...]): Scopes de autenticación requeridos para acceder a este prompt (por defecto: tupla vacía)

Propiedades:

• name: El nombre de la función (derivado de func.__name__)

• title: Un título legible por humanos (derivado del nombre de la función)

• description: El docstring de la función

• arguments: Tupla de objetos PromptArgument que definen los argumentos del prompt

Clase Arguments

La clase Arguments se pasa a las funciones de herramientas y prompts para proporcionar acceso a entradas, solicitud y estado.

Parámetros:

• request: El objeto Request de Starlette

• inputs: Los datos de entrada/argumento validados (el tipo depende de la definición de la herramienta/prompt)

Métodos:

• get_state_key(key: str, _object_type: type[TKey]) -> TKey: Accede a un valor del estado del ciclo de vida. Lanza ServerError si la clave no existe.

Clase NoArguments

Un modelo Pydantic vacío que se puede usar como una alternativa más clara a type(None) al definir herramientas o prompts sin argumentos.

from http_mcp.types import NoArguments

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

Autorización OAuth 2.1 (auth_mcp)

El paquete auth_mcp añade autorización OAuth 2.1 conforme a estándares a tu servidor MCP. Instala con el extra auth:

pip install http-mcp[auth]

Inicio rápido

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)

Esto te proporciona:

• Validación de tokens Bearer en todos los endpoints de MCP (seguro por defecto)

• Endpoint de descubrimiento /.well-known/oauth-protected-resource (RFC 9728)

• Cabeceras WWW-Authenticate en 401/403 con el parámetro resource_metadata

• Cabeceras de seguridad (HSTS, nosniff, no-store)

• Middleware personalizado opcional mediante el parámetro middlewares

Para documentación completa, mejores prácticas y detalles de la superficie de seguridad, consulta README de auth_mcp.

Superficies de seguridad por endpoint

POST /mcp — Endpoint JSON-RPC de MCP

• Autenticación — Cuando se usa auth_mcp, los tokens Bearer se extraen de la cabecera Authorization y se validan mediante TokenValidator. Los tokens que superen los 2048 caracteres o que contengan caracteres fuera del patrón b64token de RFC 6750 se rechazan antes de llegar al validador. Sin auth_mcp, la autenticación la maneja el AuthenticationMiddleware de Starlette.

• Autorización — Filtrado basado en scopes mediante has_required_scope() de Starlette. Las herramientas y los prompts sin scopes coincidentes se ocultan de los listados y se bloquean en la invocación. La validación de cabeceras de solicitud resuelve los esquemas de herramientas mediante la misma comprobación de scopes, por lo que un llamador al que se le oculta una herramienta no puede conocer sus argumentos x-mcp-header a partir de un mensaje de discrepancia.

• Validación de entrada — Los mensajes JSON-RPC se validan con Pydantic. El cuerpo de la solicitud está limitado a 4 MB, aplicado durante la lectura: un Content-Length sobredimensionado se rechaza antes de que se lea el cuerpo, y un cuerpo que supere el límite a mitad de la transmisión deja de almacenarse en búfer en ese punto. El Content-Type se comprueba estrictamente (solo application/json, se ignoran los parámetros de tipo de medio).

• Manejo de errores — Los nombres de herramientas y prompts se truncan a 100 caracteres en los mensajes de error. Los errores de validación de Pydantic se sanean antes de incluirlos en las respuestas.

• Cabeceras de respuesta — X-Content-Type-Options: nosniff, Cache-Control: no-store en todas las respuestas. auth_mcp añade además Strict-Transport-Security: max-age=31536000; includeSubDomains.

GET /.well-known/oauth-protected-resource — Endpoint de descubrimiento (auth_mcp)

• Autenticación — Sujeto al mismo middleware de autenticación que /mcp. Cuando require_authentication=True (por defecto), requiere un token válido. Establece False si los clientes necesitan descubrir el servidor de autorización antes de autenticarse.

• Validación de entrada — Solo se permite GET; otros métodos devuelven 405 Method Not Allowed.

• Salida — Serializado una vez al inicio desde un modelo ProtectedResourceMetadata congelado. Los campos de URI se validan como URLs HTTP/HTTPS mediante AnyHttpUrl de Pydantic.

Cabecera de respuesta WWW-Authenticate (auth_mcp)

• Inyección de cabeceras — Todos los valores de parámetros (realm, resource_metadata, scope, error, error_description) se sanean: se eliminan los caracteres CR/LF, se escapan la barra invertida y las comillas dobles según las reglas de cadenas entre comillas de RFC 7230.

• Divulgación de información — Las respuestas de error usan mensajes genéricos ("Authentication required"). Los detalles originales de AuthenticationError se descartan. Los códigos de error (invalid_token en 401) siguen RFC 6750 sin filtrar estado interno.

Transporte STDIO

• Tamaño de mensaje — Limitado a 4 MB, igual que el transporte HTTP.

• Registro — Los mensajes se truncan a 500 caracteres en los registros de depuración para evitar inundaciones de registros. Los valores de token nunca se registran.

• Cabeceras — Las cabeceras de solicitud se convierten al formato ASGI list[tuple[bytes, bytes]] adecuado.

Instalación

Requiere Python 3.12+ (usa la sintaxis de parámetros de tipo PEP 695).

Instala el paquete usando pip o uv:

pip install http-mcp

Con soporte de autorización OAuth 2.1:

pip install http-mcp[auth]

o

uv add http-mcp

Licencia

Este proyecto está licenciado bajo la Licencia MIT. Consulta el archivo LICENSE para más detalles.