Что такое MCP: открытый стандарт Anthropic для подключения инструментов

"""
Минимальный MCP-сервер: работа с файлами.
Запуск: python file_server.py
"""
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import asyncio


# Создаём сервер с именем — имя видит клиент при discovery
server = Server("file-tools")


# ── Объявляем инструменты ──

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    """Отвечаем на tools/list — возвращаем описания инструментов."""
    return [
        types.Tool(
            name="read_file",
            description="Читает содержимое файла по указанному пути",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Абсолютный или относительный путь к файлу"
                    }
                },
                "required": ["path"]
            }
        ),
        types.Tool(
            name="write_file",
            description="Записывает текст в файл (перезаписывает если существует)",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Путь к файлу"
                    },
                    "content": {
                        "type": "string",
                        "description": "Текст для записи"
                    }
                },
                "required": ["path", "content"]
            }
        ),
    ]


# ── Обрабатываем вызовы ──

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    """Отвечаем на tools/call — выполняем инструмент и возвращаем результат."""

    if name == "read_file":
        path = arguments["path"]
        try:
            with open(path, encoding="utf-8") as f:
                content = f.read()
            return [types.TextContent(type="text", text=content)]
        except FileNotFoundError:
            return [types.TextContent(type="text", text=f"Error: file not found: {path}")]
        except Exception as e:
            return [types.TextContent(type="text", text=f"Error: {e}")]

    elif name == "write_file":
        path = arguments["path"]
        content = arguments["content"]
        try:
            with open(path, "w", encoding="utf-8") as f:
                f.write(content)
            return [types.TextContent(type="text", text=f"OK: written {len(content)} chars to {path}")]
        except Exception as e:
            return [types.TextContent(type="text", text=f"Error: {e}")]

    else:
        return [types.TextContent(type="text", text=f"Unknown tool: {name}")]


# ── Запускаем сервер через stdio транспорт ──

async def main():
    async with stdio_server() as streams:
        await server.run(
            streams[0],   # stdin
            streams[1],   # stdout
            server.create_initialization_options()
        )

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

"""
MCP-клиент: подключаемся к file_server.py и вызываем инструменты.
"""
import asyncio
from mcp import ClientSession
from mcp.client.stdio import stdio_client
import mcp.types as types


async def main():
    # Запускаем сервер как subprocess и подключаемся через stdio
    async with stdio_client(
        command="python",
        args=["file_server.py"]
    ) as (read_stream, write_stream):

        async with ClientSession(read_stream, write_stream) as session:

            # Инициализация (handshake)
            await session.initialize()

            # Discovery: получаем список инструментов
            tools_response = await session.list_tools()
            print("Доступные инструменты:")
            for tool in tools_response.tools:
                print(f"  {tool.name}: {tool.description}")

            # Вызываем инструмент read_file
            result = await session.call_tool(
                "read_file",
                arguments={"path": "/etc/hostname"}
            )

            print("\nРезультат read_file:")
            for item in result.content:
                print(item.text)


asyncio.run(main())

import anthropic

client = anthropic.Anthropic()

# Подключаем MCP-сервер через Anthropic SDK
# (SDK запускает сервер как subprocess и управляет соединением)
with client.beta.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Прочитай файл /etc/hostname и скажи его содержимое"}],
    mcp_servers=[
        {
            "type": "stdio",
            "command": "python",
            "args": ["file_server.py"],
        }
    ],
    betas=["mcp-client-2025-04-04"],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

print()

# Минимальный MCP-сервер за 20 строк
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import asyncio

server = Server("my-server")

@server.list_tools()
async def list_tools():
    return [types.Tool(
        name="hello",
        description="Возвращает приветствие",
        inputSchema={"type":"object","properties":{"name":{"type":"string"}},"required":["name"]}
    )]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "hello":
        return [types.TextContent(type="text", text=f"Привет, {arguments['name']}!")]

async def main():
    async with stdio_server() as (r, w):
        await server.run(r, w, server.create_initialization_options())

asyncio.run(main())