de en es zh

Simple HTTP MCP Server

Name: Simple HTTP MCP Server
Author: yeison-liscano

by yeison-liscano

Overview Schema Related Servers Score Discussions

Python

Remote

Simple HTTP MCP Server 实现

本项目提供了一个基于 HTTP 的模型上下文协议 (MCP) 轻量级服务器实现。它允许你将 Python 函数公开为工具和提示词，并通过 JSON-RPC 接口进行远程发现和执行。它旨在与 Starlette 或 FastAPI 应用程序配合使用（参见 demo）。

特性

符合 MCP 协议：实现了用于工具和提示词发现与执行的 MCP 规范。不支持通知。
HTTP 和 STDIO 传输：使用 HTTP (POST 请求) 或 STDIO 进行通信。
异步支持：基于 Starlette 或 FastAPI 构建，用于异步请求处理。
类型安全：利用 Pydantic 进行稳健的数据验证和序列化。
服务器状态管理：通过 get_state_key 方法在生命周期上下文中访问共享状态。
请求访问：从你的工具和提示词中访问传入的请求对象。
授权作用域：支持使用 Starlette 的身份验证系统进行基于作用域的授权。
错误处理：工具可以选择返回错误消息，而不是抛出异常。
OAuth 2.1 授权：可选的 auth_mcp 包，支持 Bearer 令牌验证、受保护资源元数据 (RFC 9728) 和 WWW-Authenticate 错误响应。使用 pip install http-mcp[auth] 安装。

服务器架构

该库提供了一个单一的 MCPServer 类，它使用生命周期 (lifespan) 来管理整个应用程序生命周期内的共享状态。

MCPServer

MCPServer 旨在与 Starlette 的生命周期系统配合使用，以管理共享的服务器状态。

关键特性：

基于生命周期：使用 Starlette 的生命周期事件来初始化和管理共享服务器状态
应用级状态：状态在整个应用程序生命周期内持久存在，而非按请求持久化
灵活：可与存储在生命周期状态中的任何自定义上下文类一起使用

构造函数参数：

name (str): 你的 MCP 服务器名称
version (str): 你的 MCP 服务器版本
tools (tuple[Tool, ...]): 要公开的工具元组（默认：空元组）
prompts (tuple[Prompt, ...]): 要公开的提示词元组（默认：空元组）
instructions (str | None): 可选的 AI 助手使用说明

使用示例：

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)

工具

工具是可以由客户端调用的函数。

基础工具示例

定义工具的参数和输出：

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

定义工具：

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

实例化服务器：

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

无参数工具

你可以定义不需要任何输入参数的工具：

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

或者，你可以使用 NoArguments 类以获得更好的清晰度：

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

带错误处理的工具

工具可以选择返回错误消息，而不是抛出异常：

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

当 return_error_message=True 时，工具将返回一个包含错误详细信息的 ErrorMessage 模型，而不是抛出 ToolInvocationError。

带授权作用域的工具

你可以根据身份验证作用域限制工具访问：

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

注意：你需要为 Starlette 应用设置身份验证中间件，作用域才能正常工作。Tool 上的 scopes 字段是主要的授权门控——框架会在调用前按作用域过滤工具。上述工具函数内部的 raise ToolInvocationError(...) 调用是可选的深度防御检查，它会向客户端返回适当的错误响应，而不是静默失败。

服务器状态管理

服务器使用 Starlette 的生命周期系统来管理整个应用程序生命周期内的共享状态。状态在应用程序启动时初始化，并持续到关闭。上下文通过 Arguments 对象上的 get_state_key 方法访问。

这对于在工具之间共享资源（如数据库连接池、HTTP 客户端、缓存或任何应用程序状态）非常有用。

数据库连接池

最常见的模式——在启动时初始化连接池，在所有工具中共享它，并在关闭时将其关闭：

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

共享 HTTP 客户端

在工具之间共享单个 httpx.AsyncClient 以重用连接，并一次性配置基础 URL、标头或超时：

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

内存缓存

在同一服务器生命周期内的工具调用之间共享可变状态（如缓存或计数器）：

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

所有共享同一 AppContext 实例的工具都能立即看到彼此的写入，因为生命周期产生了一个单一的共享对象。

注意：普通的 dict 和 int 不是线程安全的。如果你的工具并发运行（例如，通过线程调度的同步工具），请使用 asyncio.Lock 或线程安全的数据结构来保护共享的可变状态。

请求访问

你可以从工具中访问传入的请求对象。请求对象被传递给每个工具调用，可用于访问标头、Cookie 和其他请求数据（例如 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,),
)

提示词

你可以添加由用户选择调用的交互式模板。提示词现在支持生命周期状态访问，类似于工具。

基础提示词示例

定义提示词的参数：

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

实例化服务器：

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

无参数提示词

你可以定义不需要任何输入参数的提示词：

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

或者，你可以使用 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,
    ),
)

带生命周期状态的提示词

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

带授权作用域的提示词

你可以根据身份验证作用域限制提示词访问：

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

注意：你需要为 Starlette 应用设置身份验证中间件，作用域才能正常工作。

STDIO 传输

除了 HTTP 传输外，服务器还支持用于通信的 STDIO 传输。这对于通过标准输入/输出进行通信的命令行应用程序和集成非常有用。

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

request_headers 参数允许你传递将包含在请求上下文中的标头，即使在使用 STDIO 传输时也能启用身份验证和其他基于标头的功能。

身份验证与授权

该库与 Starlette 的身份验证系统集成，为工具和提示词提供基于作用域的授权。

设置身份验证中间件

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)

作用域的工作原理

身份验证中间件：中间件验证每个请求，并通过 AuthCredentials 将作用域分配给用户。
工具/提示词作用域：定义工具或提示词时，可以使用 scopes 参数指定所需的作用域。
访问控制：服务器会自动根据用户授予的作用域过滤工具和提示词。没有所需作用域的工具和提示词在列表中不可见，也无法调用。
多重作用域：如果你指定了多个作用域（例如 scopes=("admin", "superuser")），用户至少需要其中一个作用域才能访问该工具或提示词。

API 参考

Tool 类

Tool 类用于定义可由客户端调用的工具。

参数：

func: 要调用的函数。可以是同步或异步的。函数可以：
- 接受一个 Arguments[TInputs] 参数
- 不接受任何参数
inputs: 用于输入验证的 Pydantic 模型类。对于没有输入的工具，使用 type(None) 或 NoArguments
output: 用于输出验证的 Pydantic 模型类
return_error_message (bool): 如果为 True，工具错误将返回 ErrorMessage 而不是抛出异常（默认：False）
scopes (tuple[str, ...]): 访问此工具所需的身份验证作用域（默认：空元组）

属性：

name: 函数名称（派生自 func.__name__）
title: 人类可读的标题（派生自函数名称）
description: 函数的文档字符串
input_schema: 输入参数的 JSON 模式
output_schema: 输出的 JSON 模式

Prompt 类

Prompt 类用于定义可由客户端调用的提示词。

参数：

func: 要调用的函数。可以是同步或异步的。函数可以：
- 接受一个 Arguments[TArguments] 参数
- 不接受任何参数
- 必须返回 tuple[PromptMessage, ...]
arguments_type: 用于参数验证的 Pydantic 模型类。对于没有参数的提示词，使用 type(None) 或 NoArguments
scopes (tuple[str, ...]): 访问此提示词所需的身份验证作用域（默认：空元组）

属性：

name: 函数名称（派生自 func.__name__）
title: 人类可读的标题（派生自函数名称）
description: 函数的文档字符串
arguments: 定义提示词参数的 PromptArgument 对象元组

Arguments 类

Arguments 类被传递给工具和提示词函数，以提供对输入、请求和状态的访问。

参数：

request: Starlette Request 对象
inputs: 经过验证的输入/参数数据（类型取决于工具/提示词定义）

方法：

get_state_key(key: str, _object_type: type[TKey]) -> TKey: 从生命周期状态访问值。如果键不存在，则引发 ServerError。

NoArguments 类

一个空的 Pydantic 模型，在定义没有参数的工具或提示词时，可以用作 type(None) 的更清晰替代方案。

from http_mcp.types import NoArguments

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

OAuth 2.1 授权 (auth_mcp)

auth_mcp 包为你的 MCP 服务器添加了符合标准的 OAuth 2.1 授权。使用 auth 额外项安装：

pip install http-mcp[auth]

快速入门

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)

这为你提供了：

所有 MCP 端点上的 Bearer 令牌验证（默认安全）
/.well-known/oauth-protected-resource 发现端点 (RFC 9728)
带有 resource_metadata 参数的 401/403 上的 WWW-Authenticate 标头
安全标头 (HSTS, nosniff, no-store)
通过 middlewares 参数的可选自定义中间件

有关完整文档、最佳实践和安全面详细信息，请参阅 auth_mcp README。

各端点的安全面

`POST /mcp` — MCP JSON-RPC 端点

身份验证 — 使用 auth_mcp 时，Bearer 令牌从 Authorization 标头中提取，并通过 TokenValidator 进行验证。超过 2048 个字符或包含 RFC 6750 b64token 模式之外字符的令牌在到达验证器之前会被拒绝。如果没有 auth_mcp，身份验证由 Starlette 的 AuthenticationMiddleware 处理。
授权 — 通过 Starlette 的 has_required_scope() 进行基于作用域的过滤。没有匹配作用域的工具和提示词在列表中被隐藏，并在调用时被阻止。
输入验证 — JSON-RPC 消息由 Pydantic 验证。请求体上限为 4 MB。严格检查 Content-Type（仅限 application/json，忽略媒体类型参数）。
错误处理 — 错误消息中的工具和提示词名称截断为 100 个字符。Pydantic 验证错误在包含在响应中之前会被清理。
响应标头 — 所有响应上的 X-Content-Type-Options: nosniff, Cache-Control: no-store。auth_mcp 额外添加了 Strict-Transport-Security: max-age=31536000; includeSubDomains。

`GET /.well-known/oauth-protected-resource` — 发现端点 (auth_mcp)

身份验证 — 受与 /mcp 相同的身份验证中间件约束。当 require_authentication=True（默认）时，需要有效的令牌。如果客户端需要在身份验证前发现授权服务器，请设置为 False。
输入验证 — 仅允许 GET；其他方法返回 405 Method Not Allowed。
输出 — 在启动时从冻结的 ProtectedResourceMetadata 模型序列化一次。URI 字段通过 Pydantic 的 AnyHttpUrl 验证为 HTTP/HTTPS URL。

`WWW-Authenticate` 响应标头 (auth_mcp)

标头注入 — 所有参数值（realm, resource_metadata, scope, error, error_description）均已清理：根据 RFC 7230 引号字符串规则剥离 CR/LF 字符，并转义反斜杠和双引号。
信息泄露 — 错误响应使用通用消息（"Authentication required"）。原始的 AuthenticationError 详细信息被丢弃。错误代码（401 上的 invalid_token）遵循 RFC 6750，不会泄露内部状态。

STDIO 传输

消息大小 — 上限为 4 MB，与 HTTP 传输匹配。
日志记录 — 调试日志中的消息截断为 500 个字符，以防止日志泛滥。令牌值永远不会被记录。
标头 — 请求标头转换为正确的 ASGI list[tuple[bytes, bytes]] 格式。

安装

需要 Python 3.12+（使用 PEP 695 类型参数语法）。

使用 pip 或 uv 安装包：

pip install http-mcp

支持 OAuth 2.1 授权：

pip install http-mcp[auth]

或

uv add http-mcp

许可证

本项目采用 MIT 许可证。详情请参阅 LICENSE 文件

Install Server

security – no known vulnerabilities

license - permissive license

quality - not tested

How are these scores calculated?

Resources

GitHub Repository

Need Help?

Reddit Discussion

Related Servers

Unclaimed servers have limited discoverability.

Looking for Admin?

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

Tools

Appeared in Searches

Simple MCP example using HTTP transport

Latest Blog Posts

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

MCP directory API

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

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

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

Simple HTTP MCP Server 实现

目录

特性

服务器架构

MCPServer

工具

基础工具示例

无参数工具

带错误处理的工具

带授权作用域的工具

服务器状态管理

数据库连接池

共享 HTTP 客户端

内存缓存

请求访问

提示词

基础提示词示例

无参数提示词

带生命周期状态的提示词

带授权作用域的提示词

STDIO 传输

使用 STDIO 传输

身份验证与授权

设置身份验证中间件

作用域的工作原理

API 参考

Tool 类

Prompt 类

Arguments 类

NoArguments 类

OAuth 2.1 授权 (auth_mcp)

快速入门

各端点的安全面

POST /mcp — MCP JSON-RPC 端点

GET /.well-known/oauth-protected-resource — 发现端点 (auth_mcp)

WWW-Authenticate 响应标头 (auth_mcp)

STDIO 传输

安装

许可证

Resources

Looking for Admin?

Tools

Appeared in Searches

Latest Blog Posts

MCP directory API

`POST /mcp` — MCP JSON-RPC 端点

`GET /.well-known/oauth-protected-resource` — 发现端点 (auth_mcp)

`WWW-Authenticate` 响应标头 (auth_mcp)