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

Соберите инструмент get_artifact, который читает одну запись inventory. HTTP-клиент отвечает за запрос и разбор ответа, MCP — за аргументы и контракт. Все ответы в этой практике синтетические.

**В результате:** Три полных файла: HTTP-клиент, тонкий MCP tool и проверка tools/list → tools/call на настоящем SDK, без сети.

**Перед началом:** Python 3.11, первый рабочий MCP tool и понимание входной схемы. Начните с уроков о первом сервере и валидации.

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

```text
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
```

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

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

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

### http_api_client.py

```python
"""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.py

```python
"""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.py

```python
"""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
```

```text
{"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 настоящим, выполните следующую практику с семью тестами. После замены адреса, модели или авторизации эти тесты нужно дополнить примерами именно вашего контракта.

## Версия и документация

Python 3.11; mcp==1.30.0; httpx==0.28.1; pydantic==2.13.5. Проверка выполняет MCP-вызовы в памяти и HTTP через MockTransport. Не проверены внешний API, удалённый MCP и установка новой среды. Документация прочитана 12 октября 2026 года.

- [MCP: схема, результат и ошибки инструмента](https://modelcontextprotocol.io/specification/2025-11-25/server/tools) · Протокол 2025-11-25

- [Python SDK v1: тестирование в памяти](https://py.sdk.modelcontextprotocol.io/v1/testing/) · Пример проверен на mcp==1.30.0

- [HTTPX: MockTransport](https://www.python-httpx.org/advanced/transports/#mock-transports) · Пример проверен на httpx==0.28.1

- [HTTPX: connect, read, write и pool timeout](https://www.python-httpx.org/advanced/timeouts/) · Документация сверена 12.10.2026

- [HTTPX: исключения запроса](https://www.python-httpx.org/exceptions/) · Документация сверена 12.10.2026

- [Python 3.11: asyncio.timeout](https://docs.python.org/3.11/library/asyncio-task.html#asyncio.timeout) · Python 3.11