de en es ja ko ru zh

Simple HTTP MCP Server

Name: Simple HTTP MCP Server
Author: yeison-liscano

by yeison-liscano

Overview Schema Related Servers Score Discussions

Python

Remote

Простая реализация HTTP MCP-сервера

Этот проект предоставляет легковесную реализацию сервера для протокола Model Context Protocol (MCP) через HTTP. Он позволяет предоставлять функции Python в виде инструментов и промптов, которые могут быть обнаружены и выполнены удаленно через JSON-RPC интерфейс. Предназначен для использования с приложением Starlette или FastAPI (см. демо).

Возможности

Соответствие протоколу MCP: Реализует спецификацию MCP для обнаружения и выполнения инструментов и промптов. Уведомления не поддерживаются.
Одна ревизия протокола: Поддерживает только stateless-ревизию 2026-07-28 — server/discover, _meta на каждый запрос, без handshake и без сессии. Единый путь диспетчеризации означает, что запрос не может выбрать более слабую обработку, объявив более старую ревизию.
Транспорт HTTP и STDIO: Использует HTTP (POST-запросы) или STDIO для связи.
Поддержка асинхронности: Построен на Starlette или FastAPI для асинхронной обработки запросов.
Типобезопасность: Использует Pydantic для надежной проверки данных и сериализации.
Управление состоянием сервера: Доступ к общему состоянию через контекст lifespan с помощью метода get_state_key.
Доступ к запросу: Доступ к объекту входящего запроса из ваших инструментов и промптов.
Области авторизации: Поддержка авторизации на основе областей с использованием системы аутентификации Starlette.
Обработка ошибок: Инструменты могут опционально возвращать сообщения об ошибках вместо выбрасывания исключений.
Авторизация OAuth 2.1: Опциональный пакет auth_mcp с проверкой Bearer-токенов, метаданными защищенного ресурса (RFC 9728) и ответами об ошибках WWW-Authenticate. Установка: pip install http-mcp[auth].

Архитектура сервера

Библиотека предоставляет единый класс MCPServer, который использует lifespan для управления общим состоянием на протяжении всего жизненного цикла приложения.

MCPServer

Класс MCPServer предназначен для работы с системой lifespan в Starlette для управления общим состоянием сервера.

Ключевые характеристики:

На основе lifespan: Использует события lifespan в Starlette для инициализации и управления общим состоянием сервера
Состояние на уровне приложения: Состояние сохраняется на протяжении всего жизненного цикла приложения, а не на уровне отдельного запроса
Гибкость: Может использоваться с любым пользовательским классом контекста, хранящимся в состоянии lifespan

Параметры конструктора:

name (str): Имя вашего MCP-сервера
version (str): Версия вашего MCP-сервера
tools (tuple[Tool, ...]): Кортеж инструментов для предоставления (по умолчанию: пустой кортеж)
prompts (tuple[Prompt, ...]): Кортеж промптов для предоставления (по умолчанию: пустой кортеж)
instructions (str | None): Необязательные инструкции для ИИ-ассистентов о том, как использовать этот сервер
cache_ttl_ms (int): Подсказка о свежести в миллисекундах, отправляемая вместе с результатами tools/list, prompts/list и server/discover (по умолчанию: 300000). Используйте 0, чтобы указать клиентам никогда не кэшировать. См. Подсказки кэширования.
cache_scope ("public" | "private" | None): Могут ли общие кэши переиспользовать эти результаты в разных контекстах авторизации. Определяется автоматически, если не задан. См. Подсказки кэширования.
allowed_origins (tuple[str, ...]): Источники (origins), которые принимает HTTP-транспорт (по умолчанию: пусто, то есть проверка отключена). См. Проверка источника.
require_origin (bool): Будет ли отклонен запрос, вообще не содержащий заголовок Origin, когда задан allowed_origins (по умолчанию: False). См. Проверка источника.

Пример использования:

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)

Версия протокола

Сервер реализует ровно одну ревизию протокола 2026-07-28, и каждый запрос проходит по одному и тому же пути. Никакого согласования версий и никакого второго набора правил, который запрос мог бы выбрать.

Критическое изменение в 0.17.0. Поддержка сессионных ревизий 2025-11-25, 2025-06-18 и 2025-03-26 была удалена вместе с initialize, notifications/initialized и ping. Клиент, который говорит только на этих ревизиях, больше не сможет общаться с этим сервером. Обслуживание одной ревизии также делает приведенные ниже заголовки метаданных запроса заслуживающими доверия: пока сосуществовали две эпохи, запрос мог пропустить проверки заголовков, объявив более старую, поэтому промежуточный узел, маршрутизирующий по Mcp-Method, мог рассинхронизироваться с сервером, действующим на основе тела запроса.

Критическое изменение в 0.18.0. ServerInterface.get_tool_input_schema теперь принимает Request, поэтому области авторизации учитываются до диспетчеризации; реализации интерфейса должны быть обновлены, тогда как пользователи MCPServer не затронуты. Отраженные значения Mcp-Param-* сравниваются текстуально, а не численно, поэтому заголовок со значением 3.0 для "replicas": 3 теперь получает -32020. Каждый метод notifications/* возвращает 404, поскольку ревизия не определяет ни одного.

Формат запроса 2026-07-28

Эта ревизия не имеет концепции сессии. На практике:

Без handshake. Каждый запрос повторно указывает версию протокола и возможности клиента в _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 и clientCapabilities обязательны; отсутствие любого из них дает -32602 и HTTP 400. Любая другая версия дает -32022, чей data.supported перечисляет единственную ревизию, на которой говорит этот сервер.
server/discover заменяет initialize для обнаружения возможностей. Он сообщает поддерживаемую версию, возможности, инструкции и идентичность сервера одним вызовом и отвечает без каких-либо предварительных запросов:
```
{
  "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"
}
```
Каждый результат содержит resultType: "complete" и блок _meta с именем сервера.
initialize, notifications/initialized, ping и logging/setLevel не существуют вместе с механизмом сессии и возобновления SSE. Они возвращают -32601 с HTTP 404. JSON-RPC-уведомление — сообщение notifications/* без id — по-прежнему получает 202 Accepted и пустое тело, поскольку JSON-RPC запрещает отвечать на них.
Обязательные заголовки запроса. Каждый POST должен отправлять MCP-Protocol-Version и Mcp-Method, а также Mcp-Name для tools/call и prompts/get. Каждый из них должен совпадать с соответствующим значением в теле, в противном случае запрос отклоняется с -32020 (HeaderMismatch) и HTTP 400 — это не дает прокси маршрутизировать по одному значению, пока сервер действует на основе другого. Значения, которые нельзя выразить в виде простого ASCII, используют оболочку =?base64?...?=, которую сервер декодирует перед сравнением.
Неизвестные инструменты и промпты сообщают -32602, как предписывают спецификации инструментов и промптов. -32002 был выведен из употребления этой ревизией.
Mcp-Session-Id и Last-Event-ID игнорируются, а GET/DELETE на MCP-эндпоинте возвращают 405 Method Not Allowed.

Многоэтапные запросы (elicitation, sampling, roots) и subscriptions/listen не реализованы: этот сервер не предоставляет возможностей, зависящих от ввода клиента, и объявляет listChanged: false, поэтому ни то, ни другое к нему не применимо.

Подсказки кэширования

Результаты tools/list, prompts/list и server/discover содержат ttlMs и cacheScope, чтобы клиенты могли избегать повторной загрузки списка, который не изменился:

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
)

Инструменты и промпты фиксируются при создании MCPServer, поэтому ttlMs на самом деле ограничивает, как долго клиент может не замечать повторное развертывание, а не то, как долго данные стабильны. Установите значение 0, чтобы попросить клиентов никогда не кэшировать.

cache_scope вычисляется автоматически, если вы его опускаете: "private", если какой-либо инструмент или промпт ограничен областью (scope) — тогда список меняется в зависимости от вызывающей стороны, и общий кэш не должен переиспользовать его в разных контекстах авторизации — и "public" в противном случае. Переопределите его, если ваше развертывание знает лучше. Обратите внимание: cacheScope управляет только кэшированием; он никогда не заменяет проверки областей для отдельных инструментов.

Проверка источника

Браузеры добавляют заголовок Origin, который позволяет серверу отклонять запросы, протаскиваемые с помощью DNS rebinding. Проверка по умолчанию выключена, чтобы существующие развертывания продолжали работать; включите ее везде, где до эндпоинта можно добраться из браузера:

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

Запрос, у которого Origin присутствует и не входит в список, получает 403 Forbidden. Запросы вообще без Origin — обычные небраузерные клиенты — по умолчанию не затрагиваются, потому что браузеры всегда отправляют Origin в POST, а модель угроз rebinding не покрывает клиентов, не являющихся браузерами.

Если эндпоинт должен обслуживать только браузерный трафик, добавьте require_origin, чтобы отклонять и запросы, опускающие этот заголовок; это сделает список разрешенных обязательным, а не рекомендательным:

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

require_origin сам по себе ничего не делает — он лишь ужесточает уже настроенный список разрешенных. При локальном запуске также привязывайтесь к 127.0.0.1, а не к 0.0.0.0.

Отражение параметров инструментов в заголовках

Инструмент может попросить клиентов копировать определенные значения аргументов в заголовки Mcp-Param-*, чтобы прокси могли маршрутизировать или ограничивать скорость на их основе, не разбирая тело. Аннотируйте поле с помощью 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")

Соответствующий клиент затем отправляет Mcp-Param-Region: us-west1 вместе с вызовом, и сервер проверяет его по телу — отклоняя запрос с -32020, если заголовок отсутствует, противоречит аргументу или отправлен, когда аргумент отсутствует. Заголовки Mcp-Param-*, на которые не претендует ни одна аннотация, игнорируются, поскольку ожидается, что промежуточные узлы будут пересылать неузнанные заголовки без изменений.

Сравнение текстуальное, со значением, как его записывает JSON: для "replicas": 3 заголовок должен читаться ровно как 3, а не 3.0, +3 или 3. Численное приведение сочло бы их равными, тогда как промежуточный узел, маршрутизирующий по исходной строке заголовка, видел бы нечто иное, и именно эту рассинхронизацию призвано предотвратить отражение.

Аннотировать можно только поля string, integer и boolean, доступные через простую цепочку свойств объекта, и никакие два поля не могут претендовать на одно и то же имя заголовка — коллизия отклоняется при создании сервера, потому что сохранение одной из двух аннотаций привело бы к тому, что другая молча не применялась бы. Не аннотируйте чувствительные значения: содержимое заголовков видно каждому промежуточному узлу на пути.

Инструменты

Инструменты — это функции, которые может вызывать клиент.

Простой пример инструмента

Определите аргументы и выходные данные для инструментов:

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

Инструменты с областями авторизации

Вы можете ограничить доступ к инструментам на основе областей аутентификации (scopes):

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, чтобы области действия работали корректно. Поле scopes в Tool является основным шлюзом авторизации — фреймворк фильтрует инструменты по области действия перед вызовом. Вызовы raise ToolInvocationError(...) внутри функций инструментов выше являются необязательными проверками защиты на случай непредвиденных обстоятельств, которые возвращают корректный ответ об ошибке клиенту вместо тихого сбоя.

Управление состоянием сервера

Сервер использует систему жизненного цикла Starlette для управления общим состоянием на протяжении всего жизненного цикла приложения. Состояние инициализируется при запуске приложения и сохраняется до его завершения. Контекст доступен через метод get_state_key объекта Arguments.

Это полезно для совместного использования таких ресурсов, как пулы подключений к базе данных, 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: Объект Request из Starlette
inputs: Проверенные входные данные/аргументы (тип зависит от определения Tool/Prompt)

Методы:

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 добавляет стандартную авторизацию OAuth 2.1 к вашему MCP-серверу. Установите с дополнительным пакетом 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)

Это дает вам:

Проверку Bearer-токенов на всех конечных точках MCP (безопасно по умолчанию)
Конечную точку обнаружения /.well-known/oauth-protected-resource (RFC 9728)
Заголовки WWW-Authenticate при 401/403 с параметром resource_metadata
Заголовки безопасности (HSTS, nosniff, no-store)
Необязательное пользовательское промежуточное ПО через параметр middlewares

Для полной документации, лучших практик и деталей поверхности безопасности см. README auth _mcp.

Поверхности безопасности по конечным точкам

`POST /mcp` — конечная точка MCP JSON-RPC

Аутентификация — При использовании auth_mcp Bearer-токены извлекаются из заголовка Authorization и проверяются через TokenValidator. Токены длиной более 2048 символов или содержащие символы вне шаблона b64token RFC 6750 отклоняются до достижения валидатора. Без auth_mcp аутентификация обрабатывается AuthenticationMiddleware из Starlette.
Авторизация — Фильтрация на основе областей через has_required_scope() из Starlette. Инструменты и промпты без соответствующих областей скрыты из списков и заблокированы при вызове. Проверка заголовков запроса разрешает схемы инструментов через ту же проверку областей, поэтому вызывающий, от которого скрыт инструмент, не может узнать его аргументы x-mcp-header из сообщения о несоответствии.
Проверка входных данных — Сообщения JSON-RPC проверяются через Pydantic. Тело запроса ограничено 4 МБ, что обеспечивается при чтении: чрезмерный Content-Length отклоняется до чтения тела, а тело, превышающее лимит в процессе передачи, перестает буферизоваться в этот момент. 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 проверяются как HTTP/HTTPS URL-адреса через AnyHttpUrl из Pydantic.

Заголовок ответа `WWW-Authenticate` (auth_mcp)

Внедрение заголовка — Все значения параметров (realm, resource_metadata, scope, error, error_description) очищаются: символы CR/LF удаляются, обратная косая черта и двойные кавычки экранируются в соответствии с правилами quoted-string RFC 7230.
Раскрытие информации — Ответы об ошибках используют общие сообщения ("Authentication required"). Детали исходного AuthenticationError отбрасываются. Коды ошибок (invalid_token при 401) следуют RFC 6750 без утечки внутреннего состояния.

Транспорт STDIO

Размер сообщения — Ограничен 4 МБ, как и 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

license - permissive license

quality

maintenance

How are these scores calculated?

Maintenance

–Maintainers

–Response time

4wRelease cycle

13Releases (12mo)

Commit activity

Issues opened vs closed

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

Related MCP Servers

JSON-RPC
Entertainment & Media
xiaobenyang-com
A
license
B
quality
D
maintenance
A Model Context Protocol (MCP) server based on OpenRPC, providing JSON-RPC function invocation and method discovery services.
Updated 7 months ago (2026-01-09 07:03 UTC)
2
1
Apache 2.0
wazza-mcp-test-server
Developer Tools Testing & QA Tools API Testing
gabrielribeirogpx-arch
F
license
Not graded
quality
B
maintenance
A simple HTTP server to validate MCP infrastructure for the Wazza MCP client, exposing endpoints for tool discovery and calling.
Updated 2 months ago (2026-06-20 04:19 UTC)
remote-mcp
Developer Tools AI & Machine Learning
mo3tssem
F
license
Not graded
quality
D
maintenance
Enables building and running MCP servers over streamable HTTP, exposing tools to AI assistants like Cursor, with examples of mounting multiple servers in FastAPI.
Updated a year ago (2025-09-25 10:17 UTC)
mcp-app
Developer Tools App Automation
echomodel
F
license
Not graded
quality
C
maintenance
Framework for building and running MCP servers as HTTP services. Define tools as pure Python functions, wire up with two lines, run with one command.
Updated 2 months ago (2026-06-19 19:28 UTC)

View all related MCP servers

Related MCP Connectors

secureflows-mcp-server
MCP server for secureFlows: token-free URL builders and integration-linting tools for AI agents.
Backtest360
MCP server exposing the Backtest360 engine API as tools for AI agents.
clipkit
MCP server for Clipkit — gives AI agents a video toolbox via the Clipkit schema.

View all MCP Connectors

Appeared in Searches

Simple MCP example using HTTP transport

Latest Blog Posts

Who's Calling? MCP Hosts Are an Identity Blind Spot (And the Spec Knows It)
By Om-Shree-0709 on July 25, 2026.
mcp
Agent Identity
OAuth 2.1
Your AI Chatbot Just Exposed Your CEO's Salary to an Intern
By Om-Shree-0709 on July 2, 2026.
Agent Identity
MCP Security
OAuth Delegation
Why MCP Servers Need Execution Sandboxing (And Why Your Current Stack Isn't Enough)
By Om-Shree-0709 on June 30, 2026.
Agentic Ai
Prompt Injection
WebAssembly

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

Простая реализация HTTP MCP-сервера

Оглавление

Возможности

Архитектура сервера

MCPServer

Версия протокола

Формат запроса 2026-07-28

Подсказки кэширования

Проверка источника

Отражение параметров инструментов в заголовках

Инструменты

Простой пример инструмента

Инструменты без аргументов

Инструменты с обработкой ошибок

Инструменты с областями авторизации

Управление состоянием сервера

Пул подключений к базе данных

Общий 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

Установка

Лицензия

Maintenance

Resources

Looking for Admin?

Tools

Related MCP Servers

JSON-RPC

wazza-mcp-test-server

remote-mcp

mcp-app

Related MCP Connectors

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)