Simple HTTP MCP Server
Implementación de Servidor MCP HTTP Simple
Este proyecto proporciona una implementación de servidor ligera para el Protocolo de Contexto de Modelo (MCP) sobre HTTP. Le permite exponer funciones de Python como herramientas y prompts que pueden ser descubiertos y ejecutados de forma remota a través de una interfaz JSON-RPC. Está diseñado para ser utilizado con una aplicación Starlette o FastAPI (ver demo).
Tabla de Contenidos
Related MCP server: Python REPL MCP Server
Características
Compatible con el Protocolo MCP: Implementa la especificación MCP para el descubrimiento y ejecución de herramientas y prompts. No hay soporte para notificaciones.
Transporte HTTP y STDIO: Utiliza HTTP (solicitudes POST) o STDIO para la comunicación.
Soporte Asíncrono: Construido sobre
StarletteoFastAPIpara el manejo de solicitudes asíncronas.Seguridad de Tipos: Aprovecha
Pydanticpara una validación y serialización de datos robusta.Gestión del Estado del Servidor: Acceda al estado compartido a través del contexto de ciclo de vida utilizando 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 utilizando el sistema de autenticación de Starlette.
Manejo de Errores: Las herramientas pueden devolver opcionalmente mensajes de error en lugar de lanzar excepciones.
Autorización OAuth 2.1: Paquete opcional
auth_mcpcon validación de token Bearer, Metadatos de Recurso Protegido (RFC 9728) y respuestas de errorWWW-Authenticate. Instale conpip install http-mcp[auth].
Arquitectura del Servidor
La biblioteca proporciona una única clase MCPServer que utiliza el ciclo de vida (lifespan) para gestionar el estado compartido a lo largo de 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 utilizarse 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 MCPversion(str): La versión de su servidor MCPtools(tuple[Tool, ...]): Tupla de herramientas a exponer (predeterminado: tupla vacía)prompts(tuple[Prompt, ...]): Tupla de prompts a exponer (predeterminado: tupla vacía)instructions(str | None): Instrucciones opcionales para asistentes de IA sobre cómo usar este servidor
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)Herramientas
Las herramientas son las funciones que pueden ser llamadas por el cliente.
Ejemplo Básico de Herramienta
Defina los argumentos y la salida para 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.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"]
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 Manejo 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: Necesita configurar el middleware de autenticación en su aplicación Starlette para que los ámbitos funcionen correctamente. El campo scopes en Tool es la puerta de autorización principal: el framework filtra las herramientas por ámbito antes de la invocación. Las llamadas a raise ToolInvocationError(...) dentro de las funciones de herramienta anteriores son comprobaciones de defensa en profundidad opcionales 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 comienza 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 grupos de conexiones a bases de datos, clientes HTTP, cachés o cualquier estado de la aplicación entre herramientas.
Grupo de Conexiones a Base de Datos
El patrón más común: inicializar un grupo 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
Comparta un único httpx.AsyncClient entre herramientas para reutilizar conexiones y configurar URLs base, encabezados 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
Comparta 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: Los dict e int simples no son seguros para hilos. Si sus herramientas se ejecutan simultáneamente (por ejemplo, herramientas síncronas despachadas a través de hilos), proteja el estado mutable compartido con un asyncio.Lock o utilice estructuras de datos seguras para hilos.
Acceso a la Solicitud
Puede acceder al objeto de solicitud entrante desde sus herramientas. El objeto de solicitud se pasa a cada llamada de herramienta y puede utilizarse para acceder a encabezados, 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
Puede añadir plantillas interactivas que se invocan por elección del usuario. Los prompts ahora admiten el acceso al estado del ciclo de vida, similar a las herramientas.
Ejemplo Básico de Prompt
Defina 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,
),
)Instancie 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
Puede 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, puede 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 de 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 Ámbitos de Autorización
Puede restringir el acceso a los prompts según los ámbitos 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: Necesita configurar el middleware de autenticación en su aplicación Starlette para que los ámbitos 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 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 le permite pasar encabezados que se incluirán en el contexto de la solicitud, permitiendo la autenticación y otras características basadas en encabezados 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 ámbitos 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 Ámbitos
Middleware de Autenticación: El middleware autentica cada solicitud y asigna ámbitos al usuario a través de
AuthCredentials.Ámbitos de Herramienta/Prompt: Al definir herramientas o prompts, puede especificar los ámbitos requeridos utilizando el parámetro
scopes.Control de Acceso: El servidor filtra automáticamente las herramientas y prompts según los ámbitos concedidos al usuario. Las herramientas y prompts sin los ámbitos requeridos no son visibles en los listados y no pueden ser invocados.
Múltiples Ámbitos: Si especifica múltiples ámbitos (por ejemplo,
scopes=("admin", "superuser")), el usuario necesita al menos uno de esos ámbitos para acceder a la herramienta o 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. Usetype(None)oNoArgumentspara herramientas sin entradasoutput: La clase de modelo Pydantic para la validación de salidareturn_error_message(bool): Si esTrue, los errores de la herramienta devuelvenErrorMessageen lugar de lanzar excepciones (predeterminado:False)scopes(tuple[str, ...]): Ámbitos de autenticación requeridos para acceder a esta herramienta (predeterminado: tupla vacía)
Propiedades:
name: El nombre de la función (derivado defunc.__name__)title: Un título legible por humanos (derivado del nombre de la función)description: El docstring de la funcióninput_schema: Esquema JSON para los parámetros de entradaoutput_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. Usetype(None)oNoArgumentspara prompts sin argumentosscopes(tuple[str, ...]): Ámbitos de autenticación requeridos para acceder a este prompt (predeterminado: tupla vacía)
Propiedades:
name: El nombre de la función (derivado defunc.__name__)title: Un título legible por humanos (derivado del nombre de la función)description: El docstring de la funciónarguments: Tupla de objetosPromptArgumentque definen los argumentos del prompt
Clase Arguments
La clase Arguments se pasa a las funciones de herramienta y prompt para proporcionar acceso a las entradas, la solicitud y el estado.
Parámetros:
request: El objetoRequestde Starletteinputs: 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. LanzaServerErrorsi la clave no existe.
Clase NoArguments
Un modelo Pydantic vacío que puede utilizarse 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 los estándares a su servidor MCP. Instale 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 le proporciona:
Validación de token Bearer en todos los endpoints MCP (seguro por defecto)
Endpoint de descubrimiento
/.well-known/oauth-protected-resource(RFC 9728)Encabezados
WWW-Authenticateen 401/403 con parámetroresource_metadataEncabezados de seguridad (HSTS, nosniff, no-store)
Middleware personalizado opcional a través del parámetro
middlewares
Para obtener documentación completa, mejores prácticas y detalles de la superficie de seguridad, consulte el README de auth_mcp.
Superficies de Seguridad por Endpoint
POST /mcp — Endpoint JSON-RPC de MCP
Autenticación — Al usar
auth_mcp, los tokens Bearer se extraen del encabezadoAuthorizationy se validan medianteTokenValidator. Los tokens que exceden los 2048 caracteres o contienen caracteres fuera del patrónb64tokende RFC 6750 son rechazados antes de llegar al validador. Sinauth_mcp, la autenticación es manejada porAuthenticationMiddlewarede Starlette.Autorización — Filtrado basado en ámbitos mediante
has_required_scope()de Starlette. Las herramientas y prompts sin ámbitos coincidentes se ocultan de los listados y se bloquean al invocarlos.Validación de entrada — Mensajes JSON-RPC validados por Pydantic. Cuerpo de la solicitud limitado a 4 MB. Content-Type estrictamente verificado (solo
application/json, se ignoran los parámetros de tipo de medio).Manejo de errores — Nombres de herramientas y prompts truncados a 100 caracteres en los mensajes de error. Errores de validación de Pydantic saneados antes de su inclusión en las respuestas
Resources
Unclaimed servers have limited discoverability.
Looking for Admin?
If you are the server author, to access and configure the admin panel.
Appeared in Searches
Latest Blog Posts
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