Свой MCP tool поверх HTTP API: клиент, схема и один безопасный запрос

Проверенная учебная среда

Python 3.11 · mcp==1.30.0 · httpx==0.28.1 · pydantic==2.13.5. MCP-клиент работает в памяти, HTTP — через MockTransport, без внешнего API, модели и реального ключа.

1. Зафиксируйте небольшой контракт

Учебный inventory хранит оборудование. Вход — item_id вида KIT-42. Выход — item_id, title и status: ready или maintenance. Это выдуманный сервис на домене .invalid, а не действующий поставщик. Мы читаем одну запись GET-запросом; операций записи и персональных данных здесь нет.

Слой	Его задача	Чем проверяем
HTTP-клиент	Адрес, заголовок, предел ожидания, статус и форма JSON	MockTransport и тесты отказов
MCP tool	Имя, схема аргументов и структурированный результат	Настоящий SDK-клиент в памяти
Ваш реальный сервис	Сеть, права доступа, договорённости API	Отдельная проверка после адаптации

Не принимайте URL от пользователя вместо item_id. Фиксированный адрес и узкий идентификатор не дают превратить этот tool в произвольный HTTP-прокси. Клиент ниже повторно проверяет идентификатор, потому что его можно вызвать и без MCP.

2. Подготовьте отдельную папку

Сохраните все файлы этой практики в одной новой папке. Не заменяйте файлы первой лаборатории. Используйте отдельное окружение Python 3.11; зависимости закреплены за проверенной версией.

requirements-http-adapter.txtText

Скачать

mcp==1.30.0
httpx==0.28.1
pydantic==2.13.5

mcp==1.30.0
httpx==0.28.1
pydantic==2.13.5

PowerShell

py -3.11 -m venv .venv
.\.venv\Scripts\python.exe -m pip install -r requirements-http-adapter.txt

py -3.11 -m venv .venv
.\.venv\Scripts\python.exe -m pip install -r requirements-http-adapter.txt

Эти команды предназначены для вашей новой учебной папки. Автор проверил код в уже установленной среде и не выполнял новую установку. Возможность установки пакетов на другом компьютере не входит в сохранённый результат.

3. Отделите HTTP-клиент от MCP

Сохраните http_api_client.py. В него передаётся AsyncClient, поэтому тест выбирает транспорт, а вызывающий код закрывает клиент через async with. Нет глобального сетевого клиента или скрытого чтения переменных окружения.

http_api_client.pyPython

Скачать

"""Read one fictional inventory record using an injected HTTP client."""
import asyncio
import re

import httpx
from pydantic import BaseModel, ConfigDict, Field, ValidationError
from typing import Literal

ID_PATTERN = r"^[A-Z]{2,8}-[1-9][0-9]{0,5}$"
API_ROOT = "https://inventory.example.invalid/v1/"
HTTP_TIMEOUT = httpx.Timeout(5.0, connect=2.0, pool=1.0)


class Artifact(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")
    item_id: str = Field(pattern=ID_PATTERN)
    title: str = Field(min_length=1, max_length=80)
    status: Literal["ready", "maintenance"]


class AdapterError(Exception):
    """A stable public error without response bodies, URLs or credentials."""

    def __init__(self, code: str, message: str):
        self.code = code
        super().__init__(f"{code}: {message}")


class ArtifactClient:
    def __init__(self, http: httpx.AsyncClient, token: str | None = None, deadline: float = 8.0):
        if not 0 < deadline <= 60:
            raise ValueError("deadline must be between 0 and 60 seconds")
        self._http = http
        self._token = token
        self._deadline = deadline

    async def get_artifact(self, item_id: str) -> Artifact:
        """Validate the identifier, read once and return an explicit output model."""
        if not isinstance(item_id, str) or re.fullmatch(ID_PATTERN, item_id) is None:
            raise AdapterError("INVALID_ID", "Expected an identifier such as KIT-42")
        headers = {"Authorization": f"Bearer {self._token}"} if self._token else {}
        try:
            async with asyncio.timeout(self._deadline):
                response = await self._http.get(
                    API_ROOT + "items/" + item_id,
                    headers=headers,
                    timeout=HTTP_TIMEOUT,
                    follow_redirects=False,
                )
        except (TimeoutError, httpx.TimeoutException):
            raise AdapterError("TIMEOUT", "The inventory did not answer in time") from None
        except httpx.RequestError:
            raise AdapterError("NETWORK", "The inventory connection failed") from None
        status = response.status_code
        if status in (401, 403):
            raise AdapterError("AUTH", "Check inventory access outside the tool")
        if status == 404:
            raise AdapterError("NOT_FOUND", "No record with this identifier")
        if status == 429:
            raise AdapterError("RATE_LIMIT", "Request limit reached; do not retry in a loop")
        if status >= 500:
            raise AdapterError("UPSTREAM", "The inventory is temporarily unavailable")
        if status != 200:
            raise AdapterError("HTTP_STATUS", "Unexpected inventory response")
        try:
            result = Artifact.model_validate(response.json())
        except (ValueError, ValidationError):
            raise AdapterError("BAD_RESPONSE", "The inventory response has the wrong shape") from None
        if result.item_id != item_id:
            raise AdapterError("BAD_RESPONSE", "The inventory returned a different identifier")
        return result

"""Read one fictional inventory record using an injected HTTP client."""
import asyncio
import re

import httpx
from pydantic import BaseModel, ConfigDict, Field, ValidationError
from typing import Literal

ID_PATTERN = r"^[A-Z]{2,8}-[1-9][0-9]{0,5}$"
API_ROOT = "https://inventory.example.invalid/v1/"
HTTP_TIMEOUT = httpx.Timeout(5.0, connect=2.0, pool=1.0)


class Artifact(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")
    item_id: str = Field(pattern=ID_PATTERN)
    title: str = Field(min_length=1, max_length=80)
    status: Literal["ready", "maintenance"]


class AdapterError(Exception):
    """A stable public error without response bodies, URLs or credentials."""

    def __init__(self, code: str, message: str):
        self.code = code
        super().__init__(f"{code}: {message}")


class ArtifactClient:
    def __init__(self, http: httpx.AsyncClient, token: str | None = None, deadline: float = 8.0):
        if not 0 < deadline <= 60:
            raise ValueError("deadline must be between 0 and 60 seconds")
        self._http = http
        self._token = token
        self._deadline = deadline

    async def get_artifact(self, item_id: str) -> Artifact:
        """Validate the identifier, read once and return an explicit output model."""
        if not isinstance(item_id, str) or re.fullmatch(ID_PATTERN, item_id) is None:
            raise AdapterError("INVALID_ID", "Expected an identifier such as KIT-42")
        headers = {"Authorization": f"Bearer {self._token}"} if self._token else {}
        try:
            async with asyncio.timeout(self._deadline):
                response = await self._http.get(
                    API_ROOT + "items/" + item_id,
                    headers=headers,
                    timeout=HTTP_TIMEOUT,
                    follow_redirects=False,
                )
        except (TimeoutError, httpx.TimeoutException):
            raise AdapterError("TIMEOUT", "The inventory did not answer in time") from None
        except httpx.RequestError:
            raise AdapterError("NETWORK", "The inventory connection failed") from None
        status = response.status_code
        if status in (401, 403):
            raise AdapterError("AUTH", "Check inventory access outside the tool")
        if status == 404:
            raise AdapterError("NOT_FOUND", "No record with this identifier")
        if status == 429:
            raise AdapterError("RATE_LIMIT", "Request limit reached; do not retry in a loop")
        if status >= 500:
            raise AdapterError("UPSTREAM", "The inventory is temporarily unavailable")
        if status != 200:
            raise AdapterError("HTTP_STATUS", "Unexpected inventory response")
        try:
            result = Artifact.model_validate(response.json())
        except (ValueError, ValidationError):
            raise AdapterError("BAD_RESPONSE", "The inventory response has the wrong shape") from None
        if result.item_id != item_id:
            raise AdapterError("BAD_RESPONSE", "The inventory returned a different identifier")
        return result

Предел HTTPX задан отдельно для соединения, чтения, записи и ожидания свободного соединения. Read timeout ограничивает ожидание очередной порции данных, а не всю длительность запроса. Вокруг одного запроса дополнительно стоит asyncio.timeout: по умолчанию общий предел 8 секунд. Это не планировщик повторов.

Ответ проверяет строгая модель: обязательные поля, допустимый status, отсутствие неизвестных полей и совпадение item_id с запросом. Мы не передаём весь JSON поставщика ассистенту. Ошибки превращаются в короткие коды без тела ответа, адреса или токена.

4. Оставьте MCP-слой тонким

Сохраните http_api_tool.py. Аннотация item_id задаёт входную схему. Возвращаемая модель Artifact задаёт структурированный ответ. ToolError сообщает ожидаемый отказ клиенту; здесь он не заменяется успешным пустым объектом.

http_api_tool.pyPython

Скачать

"""A thin MCP tool; the caller owns the HTTP client and its lifetime."""
from typing import Annotated

from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.exceptions import ToolError
from pydantic import Field

from http_api_client import Artifact, ArtifactClient, AdapterError, ID_PATTERN


def make_server(client: ArtifactClient) -> FastMCP:
    server = FastMCP("Inventory practice")

    @server.tool()
    async def get_artifact(
        item_id: Annotated[str, Field(pattern=ID_PATTERN, strict=True)],
    ) -> Artifact:
        """Read one inventory item by an identifier such as KIT-42."""
        try:
            return await client.get_artifact(item_id)
        except AdapterError as error:
            raise ToolError(str(error)) from None

    return server

"""A thin MCP tool; the caller owns the HTTP client and its lifetime."""
from typing import Annotated

from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.exceptions import ToolError
from pydantic import Field

from http_api_client import Artifact, ArtifactClient, AdapterError, ID_PATTERN


def make_server(client: ArtifactClient) -> FastMCP:
    server = FastMCP("Inventory practice")

    @server.tool()
    async def get_artifact(
        item_id: Annotated[str, Field(pattern=ID_PATTERN, strict=True)],
    ) -> Artifact:
        """Read one inventory item by an identifier such as KIT-42."""
        try:
            return await client.get_artifact(item_id)
        except AdapterError as error:
            raise ToolError(str(error)) from None

    return server

Фабрика make_server получает готовый клиент. Не вызывайте run() для этой проверки: следующий файл соединит MCP-сервер и клиент в памяти. Это подтверждает контракт SDK, но не stdio, HTTP-транспорт MCP или авторизацию удалённого сервера.

5. Получите реальный MCP-ответ без внешнего API

Сохраните check_http_adapter.py и запустите его. MockTransport подставляет JSON для KIT-42 и 404 для KIT-43. Проверка также посылает неверный идентификатор и убеждается, что HTTP-запрос не появился.

check_http_adapter.pyPython

Скачать

"""Exercise real SDK discovery and calls with an entirely synthetic HTTP transport."""
import asyncio
from datetime import timedelta
import json

import httpx
from mcp.shared.memory import create_connected_server_and_client_session

from http_api_client import ArtifactClient, ID_PATTERN
from http_api_tool import make_server


async def main():
    requests = []

    def handler(request: httpx.Request) -> httpx.Response:
        requests.append(str(request.url))
        if request.url.path.endswith("KIT-43"):
            return httpx.Response(404, text="private upstream diagnostic")
        return httpx.Response(200, json={"item_id": "KIT-42", "title": "Demo kit", "status": "ready"})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
        server = make_server(ArtifactClient(http))
        async with create_connected_server_and_client_session(
            server, read_timeout_seconds=timedelta(seconds=5), raise_exceptions=True
        ) as session:
            tools = await session.list_tools()
            tool = next(tool for tool in tools.tools if tool.name == "get_artifact")
            assert tool.inputSchema["properties"]["item_id"]["pattern"] == ID_PATTERN
            assert "item_id" in tool.inputSchema["required"]
            assert tool.outputSchema is not None
            assert {"item_id", "title", "status"} <= set(tool.outputSchema["required"])
            success = await session.call_tool("get_artifact", {"item_id": "KIT-42"})
            assert not success.isError
            assert success.structuredContent == {"item_id": "KIT-42", "title": "Demo kit", "status": "ready"}
            missing = await session.call_tool("get_artifact", {"item_id": "KIT-43"})
            assert missing.isError
            assert "NOT_FOUND" in missing.content[0].text
            before = len(requests)
            invalid = await session.call_tool("get_artifact", {"item_id": "../secret"})
            assert invalid.isError and len(requests) == before
            print(json.dumps({"tool": tool.name, "success": success.structuredContent, "not_found_is_error": missing.isError, "invalid_input_http_requests": len(requests) - before}, ensure_ascii=False))
    assert http.is_closed
    print("PASS: real SDK list/call, mocked HTTP, client closed")


if __name__ == "__main__":
    asyncio.run(main())

"""Exercise real SDK discovery and calls with an entirely synthetic HTTP transport."""
import asyncio
from datetime import timedelta
import json

import httpx
from mcp.shared.memory import create_connected_server_and_client_session

from http_api_client import ArtifactClient, ID_PATTERN
from http_api_tool import make_server


async def main():
    requests = []

    def handler(request: httpx.Request) -> httpx.Response:
        requests.append(str(request.url))
        if request.url.path.endswith("KIT-43"):
            return httpx.Response(404, text="private upstream diagnostic")
        return httpx.Response(200, json={"item_id": "KIT-42", "title": "Demo kit", "status": "ready"})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
        server = make_server(ArtifactClient(http))
        async with create_connected_server_and_client_session(
            server, read_timeout_seconds=timedelta(seconds=5), raise_exceptions=True
        ) as session:
            tools = await session.list_tools()
            tool = next(tool for tool in tools.tools if tool.name == "get_artifact")
            assert tool.inputSchema["properties"]["item_id"]["pattern"] == ID_PATTERN
            assert "item_id" in tool.inputSchema["required"]
            assert tool.outputSchema is not None
            assert {"item_id", "title", "status"} <= set(tool.outputSchema["required"])
            success = await session.call_tool("get_artifact", {"item_id": "KIT-42"})
            assert not success.isError
            assert success.structuredContent == {"item_id": "KIT-42", "title": "Demo kit", "status": "ready"}
            missing = await session.call_tool("get_artifact", {"item_id": "KIT-43"})
            assert missing.isError
            assert "NOT_FOUND" in missing.content[0].text
            before = len(requests)
            invalid = await session.call_tool("get_artifact", {"item_id": "../secret"})
            assert invalid.isError and len(requests) == before
            print(json.dumps({"tool": tool.name, "success": success.structuredContent, "not_found_is_error": missing.isError, "invalid_input_http_requests": len(requests) - before}, ensure_ascii=False))
    assert http.is_closed
    print("PASS: real SDK list/call, mocked HTTP, client closed")


if __name__ == "__main__":
    asyncio.run(main())

PowerShell

.\.venv\Scripts\python.exe check_http_adapter.py
$LASTEXITCODE

.\.venv\Scripts\python.exe check_http_adapter.py
$LASTEXITCODE

Ожидаемый вывод

{"tool": "get_artifact", "success": {"item_id": "KIT-42", "title": "Demo kit", "status": "ready"}, "not_found_is_error": true, "invalid_input_http_requests": 0}
PASS: real SDK list/call, mocked HTTP, client closed
0

{"tool": "get_artifact", "success": {"item_id": "KIT-42", "title": "Demo kit", "status": "ready"}, "not_found_is_error": true, "invalid_input_http_requests": 0}
PASS: real SDK list/call, mocked HTTP, client closed
0

Ожидайте этот JSON, строку PASS и код завершения 0. SDK может писать диагностические INFO в stderr — это не ответ инструмента. Сохранённая авторская проверка прошла на SDK 1.30.0; исходящие сокеты были запрещены, попыток внешнего соединения не было.

6. Разберите отказ до подключения своего API

Код	Что произошло	Следующее действие
INVALID_ID	Неверный идентификатор	Исправить аргумент; HTTP не вызывался
AUTH	401 или 403	Проверить права и ключ вне инструмента
NOT_FOUND	404	Проверить существование записи
RATE_LIMIT	429	Остановить поток запросов, изучить условия своего API
TIMEOUT / NETWORK	Не дождались ответа или оборвалось соединение	Сначала выяснить причину и проверить пределы
UPSTREAM	5xx	Не выдавать пустой ответ за найденную запись
BAD_RESPONSE	Неверный JSON или чужой item_id	Сверить контракт поставщика

Токен в ArtifactClient необязателен и используется только в Authorization. Учебная проверка не передаёт реальный ключ. В своём проекте загрузите секрет из переменной окружения вне tool и передайте клиенту: не храните его в аргументах, URL, результате или файле примера.

Клиент делает один запрос, не следует редиректам и не повторяет 429/5xx автоматически. Для реального API повторы требуют отдельного решения: какие операции допустимо повторять, сколько попыток и как учитывать Retry-After. Этот пример ничего не записывает; гарантии повторной записи и ключи идемпотентности здесь не реализованы.

Прежде чем заменить выдуманный endpoint настоящим, выполните следующую практику с семью тестами. После замены адреса, модели или авторизации эти тесты нужно дополнить примерами именно вашего контракта.

Документация, на которой основан пример

Прочитана 12 октября 2026 года. Синтетический inventory — наш учебный контракт; документация описывает MCP, HTTPX и unittest. Пример не проверяет реальный HTTP-сервис или авторизацию удалённого MCP.

MCP: схема, результат и ошибки инструмента (в новой вкладке) · Протокол 2025-11-25
Python SDK v1: тестирование в памяти (в новой вкладке) · Пример проверен на mcp==1.30.0
HTTPX: MockTransport (в новой вкладке) · Пример проверен на httpx==0.28.1
HTTPX: connect, read, write и pool timeout (в новой вкладке) · Документация сверена 12.10.2026
HTTPX: исключения запроса (в новой вкладке) · Документация сверена 12.10.2026
Python 3.11: asyncio.timeout (в новой вкладке) · Python 3.11

Вопросы по материалу

Достаточно заменить учебный адрес на адрес своего API?

Нет. Учебный клиент ожидает конкретные поля item_id, title и status, проверяет типы и совпадение идентификатора. Для другого API нужно согласовать путь, модель ответа, доступ и проверки; затем отдельно проверить реальное соединение.

Повторит ли адаптер запрос после 429 или 5xx?

Нет. В примере нет автоматических повторов: 429 превращается в RATE_LIMIT, а 5xx — в UPSTREAM. План повторов и обработку Retry-After для своего API нужно определить отдельно.