Skip to main content
Glama
yeison-liscano

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 Starlette o FastAPI para el manejo de solicitudes asíncronas.

  • Seguridad de Tipos: Aprovecha Pydantic para 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_mcp con validación de token Bearer, Metadatos de Recurso Protegido (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 (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 MCP

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

  • tools (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

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

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

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

  2. Ámbitos de Herramienta/Prompt: Al definir herramientas o prompts, puede especificar los ámbitos requeridos utilizando el parámetro scopes.

  3. 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.

  4. 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. Use 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 (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 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. Use type(None) o NoArguments para prompts sin argumentos

  • scopes (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 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 herramienta y prompt para proporcionar acceso a las entradas, la solicitud y el 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 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-Authenticate en 401/403 con parámetro resource_metadata

  • Encabezados 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 encabezado Authorization y se validan mediante TokenValidator. Los tokens que exceden los 2048 caracteres o contienen caracteres fuera del patrón b64token de RFC 6750 son rechazados antes de llegar al validador. Sin auth_mcp, la autenticación es manejada por AuthenticationMiddleware de 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

Install Server
A
security – no known vulnerabilities
A
license - permissive license
-
quality - not tested

Resources

Unclaimed servers have limited discoverability.

Looking for Admin?

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

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