Архитектура MCP: сервер, клиент, транспорты stdio и HTTP+SSE

"""
Упрощённая модель того, что делает mcp.Server под капотом.
Реальный код SDK сложнее, но принцип такой.
"""
from typing import Callable, Awaitable, Any

class MCPServerCore:
    def __init__(self, name: str):
        self.name = name
        # Словарь: метод → обработчик
        self._handlers: dict[str, Callable] = {}
        # Реестр инструментов: имя → (schema, handler_fn)
        self._tools: dict[str, tuple[dict, Callable]] = {}

    # ── Декораторы для регистрации ──

    def list_tools(self):
        """Регистрирует функцию как обработчик tools/list."""
        def decorator(fn: Callable) -> Callable:
            self._handlers["tools/list"] = fn
            return fn
        return decorator

    def call_tool(self):
        """Регистрирует диспетчер для tools/call."""
        def decorator(fn: Callable) -> Callable:
            self._handlers["tools/call"] = fn
            return fn
        return decorator

    # ── Роутер запросов ──

    async def dispatch(self, method: str, params: dict) -> Any:
        """Получает JSON-RPC метод и направляет к нужному обработчику."""
        handler = self._handlers.get(method)
        if handler is None:
            raise MethodNotFoundError(f"Unknown method: {method}")

        if method == "tools/call":
            # Для tools/call SDK передаёт name и arguments из params
            return await handler(
                name=params["name"],
                arguments=params.get("arguments", {})
            )
        else:
            return await handler()

    # ── Инициализация (handshake) ──

    async def handle_initialize(self, client_info: dict) -> dict:
        """Отвечает на initialize: возвращаем свои capabilities."""
        return {
            "protocolVersion": "2024-11-05",
            "serverInfo": {"name": self.name, "version": "1.0.0"},
            "capabilities": {
                "tools": {},      # поддерживаем tools
                "resources": {},  # поддерживаем resources
            }
        }

import asyncio, json

async def server_run_loop(read_stream, write_stream, server: MCPServerCore):
    """
    Упрощённая версия основного цикла MCP-сервера.
    read_stream: откуда читаем (stdin или HTTP body)
    write_stream: куда пишем (stdout или SSE stream)
    """
    async for raw_line in read_stream:
        # Парсим JSON-RPC запрос
        try:
            request = json.loads(raw_line)
        except json.JSONDecodeError:
            continue  # игнорируем невалидный JSON

        req_id = request.get("id")
        method = request.get("method", "")
        params = request.get("params", {})

        # Специальный случай: notifications (уведомления без id)
        # Они не требуют ответа
        if req_id is None:
            if method == "notifications/initialized":
                # клиент подтвердил инициализацию
                pass
            continue

        # Вызываем роутер и формируем ответ
        try:
            if method == "initialize":
                result = await server.handle_initialize(params)
            else:
                result = await server.dispatch(method, params)

            response = {
                "jsonrpc": "2.0",
                "id": req_id,
                "result": result
            }
        except Exception as e:
            # JSON-RPC error response
            response = {
                "jsonrpc": "2.0",
                "id": req_id,
                "error": {
                    "code": -32603,  # Internal error
                    "message": str(e)
                }
            }

        # Отправляем ответ
        line = json.dumps(response, ensure_ascii=False) + "\n"
        await write_stream.write(line.encode())

import asyncio, json, itertools
from typing import Any

class MCPClientCore:
    """Упрощённая модель MCP-клиента."""

    def __init__(self):
        self._id_counter = itertools.count(1)
        # Карта id → asyncio.Future (ожидающие ответа)
        self._pending: dict[int, asyncio.Future] = {}
        # Кэш инструментов (заполняется при discovery)
        self._tools_cache: list | None = None
        # Транспорт (устанавливается при подключении)
        self._write_stream = None

    async def send_request(self, method: str, params: dict) -> Any:
        """
        Отправляет JSON-RPC запрос и ЖДЁТ ответ.
        Неблокирующий: другие корутины могут работать пока ждём.
        """
        req_id = next(self._id_counter)
        request = {
            "jsonrpc": "2.0",
            "id": req_id,
            "method": method,
            "params": params
        }

        # Создаём Future — будет разрешён когда придёт ответ
        future: asyncio.Future = asyncio.get_event_loop().create_future()
        self._pending[req_id] = future

        # Отправляем запрос
        line = json.dumps(request, ensure_ascii=False) + "\n"
        await self._write_stream.write(line.encode())

        # Ждём ответа (не блокируем event loop)
        return await future

    async def _response_reader(self, read_stream):
        """
        Фоновая задача: читает входящие ответы и разрешает futures.
        Запускается в asyncio.create_task() при подключении.
        """
        async for raw_line in read_stream:
            try:
                msg = json.loads(raw_line)
            except json.JSONDecodeError:
                continue

            msg_id = msg.get("id")
            if msg_id in self._pending:
                future = self._pending.pop(msg_id)
                if "error" in msg:
                    future.set_exception(
                        MCPError(msg["error"]["message"], msg["error"]["code"])
                    )
                else:
                    future.set_result(msg.get("result"))

    # ── Public API ──

    async def list_tools(self) -> list:
        """Получает список инструментов (с кэшированием)."""
        if self._tools_cache is None:
            response = await self.send_request("tools/list", {})
            self._tools_cache = response.get("tools", [])
        return self._tools_cache

    async def call_tool(self, name: str, arguments: dict) -> Any:
        """Вызывает инструмент на сервере."""
        return await self.send_request(
            "tools/call",
            {"name": name, "arguments": arguments}
        )

"""
Запуск MCP-сервера через stdio и вызов инструмента.
Показывает низкоуровневое взаимодействие с процессом.
"""
import asyncio
import json
from mcp import ClientSession
from mcp.client.stdio import stdio_client


async def demo_stdio():
    # stdio_client запускает server.py как subprocess
    # и создаёт пару (read_stream, write_stream) над его stdio
    async with stdio_client(
        command="python",
        args=["file_server.py"]
        # Опционально: env={"MY_VAR": "value"} — переменные окружения
        # cwd="/path/to/workdir"              — рабочая директория
    ) as (read_stream, write_stream):

        async with ClientSession(read_stream, write_stream) as session:
            # initialize: handshake с согласованием capabilities
            init_result = await session.initialize()
            print(f"Server: {init_result.serverInfo.name} v{init_result.serverInfo.version}")
            print(f"Capabilities: {init_result.capabilities}")

            # tools/list: получаем реестр инструментов
            tools = await session.list_tools()
            print(f"\nAvailable tools: {[t.name for t in tools.tools]}")

            # tools/call: вызываем конкретный инструмент
            result = await session.call_tool(
                "read_file",
                arguments={"path": "/etc/hostname"}
            )
            print(f"\nread_file result: {result.content[0].text}")

            # Сервер работает до тех пор, пока открыт context manager.
            # При выходе из 'async with stdio_client(...)' процесс завершается.

asyncio.run(demo_stdio())

"""
MCP-сервер через HTTP+SSE транспорт.
Установи: pip install mcp starlette uvicorn

Запуск: python http_server.py
Сервер будет доступен на http://localhost:8000
"""
import asyncio
import uvicorn
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp import types
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.requests import Request


# ── Логика сервера (одинаковая для любого транспорта!) ──

server = Server("http-file-tools")


@server.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="read_file",
            description="Читает файл по пути",
            inputSchema={
                "type": "object",
                "properties": {"path": {"type": "string"}},
                "required": ["path"]
            }
        ),
    ]


@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "read_file":
        try:
            with open(arguments["path"], encoding="utf-8") as f:
                return [types.TextContent(type="text", text=f.read())]
        except Exception as e:
            return [types.TextContent(type="text", text=f"Error: {e}")]
    return [types.TextContent(type="text", text=f"Unknown tool: {name}")]


# ── HTTP+SSE транспорт ──

# SseServerTransport создаёт два ASGI-handler'а
sse = SseServerTransport("/message")


async def handle_sse(request: Request):
    """GET /sse — клиент подключается и слушает события."""
    async with sse.connect_sse(
        request.scope, request.receive, request._send
    ) as streams:
        # Запускаем основной цикл сервера на этом соединении
        await server.run(
            streams[0],   # read
            streams[1],   # write
            server.create_initialization_options()
        )


async def handle_message(request: Request):
    """POST /message — клиент отправляет запросы."""
    await sse.handle_post_message(
        request.scope, request.receive, request._send
    )


# Starlette app с двумя маршрутами
app = Starlette(routes=[
    Route("/sse",     endpoint=handle_sse),
    Route("/message", endpoint=handle_message, methods=["POST"]),
])


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client


async def demo_http():
    # sse_client подключается к /sse и создаёт пару потоков
    # Требует: pip install mcp[client] httpx
    async with sse_client("http://localhost:8000/sse") as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools = await session.list_tools()
            print([t.name for t in tools.tools])

            result = await session.call_tool(
                "read_file",
                arguments={"path": "/etc/hostname"}
            )
            print(result.content[0].text)


asyncio.run(demo_http())

from mcp.server import Server
from mcp.server.models import InitializationOptions
import mcp.types as types

server = Server("my-server")

# Можно явно задать capabilities через InitializationOptions
# (по умолчанию SDK формирует их автоматически из зарегистрированных handlers)
def create_init_options() -> InitializationOptions:
    return InitializationOptions(
        server_name="my-server",
        server_version="1.0.0",
        capabilities=types.ServerCapabilities(
            tools=types.ToolsCapability(listChanged=True),
            # listChanged=True означает что сервер будет слать
            # notifications/tools/list_changed если набор инструментов изменится
            resources=types.ResourcesCapability(
                subscribe=False,
                listChanged=False
            ),
            logging=types.LoggingCapability(),
        )
    )

from mcp import types, ClientSession

async def robust_tool_call(session: ClientSession, name: str, args: dict) -> str:
    """
    Вызывает инструмент с обработкой всех трёх уровней ошибок.
    """
    try:
        result = await session.call_tool(name, args)

        # Проверяем ошибку уровня инструмента
        if result.isError:
            # Инструмент вернул ошибку — это бизнес-ошибка, не крэш
            error_text = result.content[0].text if result.content else "unknown tool error"
            return f"[TOOL ERROR] {error_text}"

        # Успешный результат
        return "\n".join(
            item.text for item in result.content
            if hasattr(item, "text")
        )

    except Exception as e:
        # Транспортная или протокольная ошибка
        # Сессия может быть в неопределённом состоянии — лучше переподключиться
        return f"[PROTOCOL/TRANSPORT ERROR] {type(e).__name__}: {e}"