Python 3.11 · mcp==1.30.0 · httpx==0.28.1 · pydantic==2.13.5. MCP-клиент работает в памяти, HTTP — через MockTransport, без внешнего API, модели и реального ключа.
1. Разделите проверки клиента и инструмента
Прямая проверка HTTP-клиента видит адрес, заголовок, число запросов и разобранную модель. MCP-проверка видит tools/list, inputSchema, outputSchema, structuredContent и isError. Если проверять только конечный JSON, можно пропустить лишний запрос, редирект или потерянную ошибку.
Все ответы создаёт MockTransport. Он не доказывает работу реального соединения и сам не воспроизводит сетевое ожидание. Для ReadTimeout тест явно возбуждает исключение. Для общего дедлайна отдельный async handler действительно ждёт до отмены — без сети и долгого сна.
2. Сохраните полный набор проверок
Сохраните test_http_adapter.py рядом с клиентом и tool. Синтетическая строка SYNTHETIC_TOKEN служит меткой: она намеренно встречается в закрытом ответе или исключении, а проверка запрещает её появление в публичной ошибке. Это не настоящий ключ.
"""Behavior checks for HTTP failures and the MCP boundary; never call a remote API."""
import asyncio
from datetime import timedelta
import unittest
import httpx
from mcp.shared.memory import create_connected_server_and_client_session
from http_api_client import AdapterError, ArtifactClient
from http_api_tool import make_server
GOOD = {"item_id": "KIT-42", "title": "Demo kit", "status": "ready"}
SYNTHETIC_TOKEN = "synthetic-secret-for-test-only"
class AdapterTests(unittest.IsolatedAsyncioTestCase):
async def test_success_request_shape_timeout_and_client_lifetime(self):
requests = []
def handler(request):
requests.append(request)
return httpx.Response(200, json=GOOD)
async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
result = await ArtifactClient(http, SYNTHETIC_TOKEN).get_artifact("KIT-42")
self.assertEqual(result.model_dump(), GOOD)
self.assertTrue(http.is_closed)
self.assertEqual(len(requests), 1)
request = requests[0]
self.assertEqual(request.method, "GET")
self.assertEqual(str(request.url), "https://inventory.example.invalid/v1/items/KIT-42")
self.assertEqual(request.headers["Authorization"], "Bearer " + SYNTHETIC_TOKEN)
self.assertNotIn(SYNTHETIC_TOKEN, str(request.url))
self.assertEqual(request.extensions["timeout"], {"connect": 2.0, "read": 5.0, "write": 5.0, "pool": 1.0})
async def test_bad_input_never_reaches_http(self):
requests = []
def handler(request):
requests.append(request)
return httpx.Response(200, json=GOOD)
async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
client = ArtifactClient(http)
for item_id in ["../secret", "https://other.invalid", "", "KIT-0", 42]:
with self.subTest(item_id=item_id), self.assertRaisesRegex(AdapterError, "INVALID_ID"):
await client.get_artifact(item_id)
self.assertEqual(requests, [])
async def test_statuses_are_safe_and_do_not_retry_or_follow_redirects(self):
for status, code in [(401, "AUTH"), (403, "AUTH"), (404, "NOT_FOUND"), (429, "RATE_LIMIT"), (500, "UPSTREAM"), (503, "UPSTREAM"), (302, "HTTP_STATUS"), (400, "HTTP_STATUS")]:
requests = []
def handler(request):
requests.append(request)
return httpx.Response(status, text=SYNTHETIC_TOKEN, headers={"Location": "https://other.invalid/", "Retry-After": "2"})
with self.subTest(status=status):
async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
with self.assertRaises(AdapterError) as caught:
await ArtifactClient(http, SYNTHETIC_TOKEN).get_artifact("KIT-42")
self.assertEqual(caught.exception.code, code)
self.assertNotIn(SYNTHETIC_TOKEN, str(caught.exception))
self.assertEqual(len(requests), 1)
async def test_timeout_and_connection_failure(self):
for exception, code in [(httpx.ReadTimeout, "TIMEOUT"), (httpx.ConnectError, "NETWORK")]:
def handler(request):
raise exception(SYNTHETIC_TOKEN, request=request)
with self.subTest(exception=exception.__name__):
async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
with self.assertRaises(AdapterError) as caught:
await ArtifactClient(http).get_artifact("KIT-42")
self.assertEqual(caught.exception.code, code)
self.assertNotIn(SYNTHETIC_TOKEN, str(caught.exception))
async def test_total_deadline_cancels_waiting_handler(self):
cancelled = asyncio.Event()
async def handler(request):
try:
await asyncio.Event().wait()
finally:
cancelled.set()
async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
with self.assertRaisesRegex(AdapterError, "TIMEOUT"):
await ArtifactClient(http, deadline=0.03).get_artifact("KIT-42")
self.assertTrue(cancelled.is_set())
async def test_invalid_json_schema_and_wrong_record(self):
responses = [httpx.Response(200, text="not JSON"), httpx.Response(200, json=[]), httpx.Response(200, json={}), httpx.Response(200, json={**GOOD, "status": "unknown"}), httpx.Response(200, json={**GOOD, "title": 42}), httpx.Response(200, json={**GOOD, "private": SYNTHETIC_TOKEN}), httpx.Response(200, json={**GOOD, "item_id": "KIT-43"})]
for response in responses:
with self.subTest(body=response.text):
async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response), trust_env=False) as http:
with self.assertRaises(AdapterError) as caught:
await ArtifactClient(http).get_artifact("KIT-42")
self.assertEqual(caught.exception.code, "BAD_RESPONSE")
self.assertNotIn(SYNTHETIC_TOKEN, str(caught.exception))
async def test_mcp_schema_success_bad_input_and_safe_failure(self):
requests = []
def handler(request):
requests.append(request)
return httpx.Response(503, text=SYNTHETIC_TOKEN) if request.url.path.endswith("KIT-43") else httpx.Response(200, json=GOOD)
async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
server = make_server(ArtifactClient(http, SYNTHETIC_TOKEN))
async with create_connected_server_and_client_session(server, read_timeout_seconds=timedelta(seconds=5), raise_exceptions=True) as session:
tool = (await session.list_tools()).tools[0]
self.assertEqual(tool.name, "get_artifact")
self.assertIn("item_id", tool.inputSchema["required"])
self.assertIsNotNone(tool.outputSchema)
success = await session.call_tool("get_artifact", {"item_id": "KIT-42"})
self.assertFalse(success.isError)
self.assertEqual(success.structuredContent, GOOD)
before = len(requests)
for arguments in [{}, {"item_id": "../secret"}, {"item_id": 42}]:
result = await session.call_tool("get_artifact", arguments)
self.assertTrue(result.isError)
self.assertEqual(len(requests), before)
failure = await session.call_tool("get_artifact", {"item_id": "KIT-43"})
self.assertTrue(failure.isError)
self.assertIsNone(failure.structuredContent)
self.assertIn("UPSTREAM", failure.content[0].text)
self.assertNotIn(SYNTHETIC_TOKEN, failure.model_dump_json())
if __name__ == "__main__":
unittest.main(verbosity=2)
.\.venv\Scripts\python.exe -m unittest -v test_http_adapter
$LASTEXITCODEtest_bad_input_never_reaches_http ... ok
test_invalid_json_schema_and_wrong_record ... ok
test_mcp_schema_success_bad_input_and_safe_failure ... ok
test_statuses_are_safe_and_do_not_retry_or_follow_redirects ... ok
test_success_request_shape_timeout_and_client_lifetime ... ok
test_timeout_and_connection_failure ... ok
test_total_deadline_cancels_waiting_handler ... ok
Ran 7 tests
OK
0В unittest имена будут дополнены названием класса, а время запуска зависит от компьютера. Важны семь ok, итог OK и код 0. INFO SDK/HTTPX в stderr допустимы. На авторской среде все семь прошли; stdout и stderr сохранены отдельно.
3. Прочитайте, что замечает каждый тест
| Тест | Что проверяет | Какой дефект поймает |
|---|---|---|
| success_request_shape… | GET, точный адрес, Authorization, настройки timeout, закрытие клиента | Токен в URL, неверный endpoint или забытое закрытие |
| bad_input… | Нет запросов при неверном item_id | Сетевой запрос до валидации |
| statuses… | 401/403/404/429/5xx/302/400 и один запрос | Лишние повторы, переход по Location, неверный код отказа |
| timeout_and_connection… | ReadTimeout и ConnectError | Потерянный отказ или утечка сырого исключения |
| total_deadline… | Реальная отмена ожидания через 0.03 s | Зависший обработчик без общего предела |
| invalid_json… | Мусор, массив, пропущенные поля, неверные типы, лишнее поле и другой id | Передача непроверенного ответа |
| mcp_schema… | Настоящие list/call, плохие аргументы, isError и безопасный результат | Неверная регистрация или успешный ответ вместо ошибки |
Тест settings timeout смотрит на параметры, переданные HTTPX; он не измеряет поведение сети. Тест дедлайна проверяет asyncio.timeout в этой программе. Проверка секрета относится к сообщению AdapterError и результату MCP; она не является аудитом всех логов, памяти или окружения процесса.
4. Убедитесь, что тест способен упасть
Сделайте отдельную копию новой учебной папки. Только в копии клиента замените код RATE_LIMIT для 429 на AUTH. Запустите unittest: test_statuses_are_safe_and_do_not_retry_or_follow_redirects должен упасть на сравнении кода, а команда завершиться ненулевым кодом. Верните RATE_LIMIT и повторите запуск.
Автор выполнил такой опыт в отдельной копии, не меняя исходный пример. Это проверяет, что отрицательные сценарии действительно различают ошибки, а не принимают любой отказ за успех. Не изменяйте ради опыта первую лабораторию или готовый рабочий проект.
5. Найдите слой проблемы
| Симптом | Что проверить сначала |
|---|---|
| ModuleNotFoundError | Все четыре .py и requirements в одной папке; запускается интерпретатор своего .venv |
| Нет inputSchema / outputSchema | Версия mcp==1.30.0, аннотации аргумента и модели Artifact |
| INVALID_ID не блокирует запрос | Прямую валидацию клиента и передачу аргументов tool |
| BAD_RESPONSE на правильной записи | Строгие типы, неизвестные поля и совпадение item_id |
| Тест deadline не заканчивается | Контекст asyncio.timeout вокруг await запроса; не увеличивать предел вслепую |
| Синтетический секрет появился в ошибке | Не возвращается ли response.text или repr исключения HTTPX/Pydantic |
Запускайте сначала отдельный упавший тест, например командой ниже. Так вы увидите отказ нужного слоя, не потеряв его в длинном выводе.
.\.venv\Scripts\python.exe -m unittest -v test_http_adapter.AdapterTests.test_statuses_are_safe_and_do_not_retry_or_follow_redirects6. Подготовьте адаптацию под свой контракт
После прохождения примера заменяйте только один слой за раз: модель ответа, путь запроса или способ передачи ключа. Добавьте синтетические fixtures своего API, прежде чем выполнять живые запросы. Доступ к реальному сервису и допустимость его использования проверяются отдельно.
Для retry сначала выберите ограниченный план повторов и поведение при 429/Retry-After. Не добавляйте while True. Для записи нужны отдельные правила повторной операции и проверка идемпотентности; текущий GET-пример их не обещает.
Следующий полезный результат — собственная таблица: «изменение контракта → какой тест падает». Если изменение полей, прав или ошибки не замечает ни одна проверка, добавьте соответствующий сценарий до расширения tool.
Документация, на которой основан пример
Прочитана 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: IsolatedAsyncioTestCase (в новой вкладке) · Python 3.11
- Python 3.11: asyncio.timeout (в новой вкладке) · Python 3.11
Вопросы по материалу
MockTransport показывает реальное поведение сети при таймауте?
Нет. В одном тесте ReadTimeout возбуждается явно. Другой тест проверяет отмену ожидающего обработчика общим дедлайном. Это проверки логики адаптера; задержки и сбои реального соединения они не воспроизводят.
Тест секрета доказывает, что ключ нигде не утечёт?
Он проверяет конкретные выходы: сообщение AdapterError и результат MCP не должны содержать синтетическую метку. Логи, память процесса и окружение этим тестом полностью не проверяются.