Как подключить OpenAI Agent SDK к серверам MCP

SDK агентов OpenAI предоставляет мощный способ для создания AI-приложений с возможностями агента. Одной из его ключевых функций является поддержка Протокола контекста модели (MCP), который стандартизирует, как приложения предоставляют контекст и инструменты для крупных языковых моделей (LLMs). Эта статья приведет вас через процесс подключения SDK агента OpenAI к серверам MCP, с детальными шагами и примером кода.

Понимание Протокола контекста модели (MCP)

MCP — это открытый протокол, который стандартизирует, как приложения предоставляют контекст для LLM. Представьте MCP как "USB-C порт" для AI-приложений — он предоставляет стандартизированный способ подключения AI-моделей к различным источникам данных и инструментам. Так же как USB-C подключает ваши устройства к различным периферийным устройствам, MCP подключает AI-модели к разным инструментам и источникам данных.

Типы серверов MCP

Спецификация MCP определяет два типа серверов на основе их транспортного механизма:

1. Серверы Stdio: Эти серверы работают как подпроцесс вашего приложения, фактически работая "локально".
2. Серверы SSE: Эти серверы работают удаленно, и вы подключаетесь к ним через URL, используя HTTP через серверные отправленные события (SSE).

SDK агентов OpenAI предоставляет соответствующие классы для обоих этих типов серверов:

• MCPServerStdio для локальных серверов stdio
• MCPServerSse для удаленных серверов SSE

Необходимые условия

Перед тем как начать, убедитесь, что у вас установлено:

Python 3.10 или выше

Установленный SDK агента OpenAI:

pip install openai-agents

Для использования локальных серверов stdio вам могут понадобиться дополнительные инструменты, такие как npx (для серверов MCP на основе JavaScript)

Подключение к локальному серверу MCP Stdio

Давайте начнем с подключения к локальному серверу MCP stdio. Мы будем использовать сервер файловой системы в качестве примера.

Шаг 1: Установка необходимых зависимостей

Если вы планируете использовать сервер файловой системы MCP, вам понадобятся Node.js и NPX:

# Установите Node.js (если еще не установлен)
# Для Ubuntu/Debian
sudo apt update
sudo apt install nodejs npm

# Для macOS
brew install node

# Проверьте установку
node --version
npx --version

Шаг 2: Настройка структуры вашего проекта

Создайте базовую структуру проекта:

my_agent_project/
├── sample_data/
│   ├── file1.txt
│   └── file2.md
├── main.py
└── README.md

Шаг 3: Подключение к серверу MCP Stdio

Вот полный пример подключения к серверу файловой системы MCP с использованием MCPServerStdio:

import asyncio
import os
import shutil
from agents import Agent, Runner, gen_trace_id, trace
from agents.mcp import MCPServerStdio

async def run_agent_with_mcp():
    # Получите путь к директории с образцами данных
    current_dir = os.path.dirname(os.path.abspath(__file__))
    sample_data_dir = os.path.join(current_dir, "sample_data")

    # Создайте и настройте MCP сервер
    async with MCPServerStdio(
        name="Сервер файловой системы",
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", sample_data_dir],
        },
    ) as mcp_server:
        # Создайте агента, который использует MCP сервер
        agent = Agent(
            name="FileAssistant",
            instructions="Используйте инструменты файловой системы для чтения и анализа файлов в директории sample_data.",
            mcp_servers=[mcp_server]
        )

        # Запустите агента с запросом от пользователя
        message = "Перечислите все файлы в директории и подведите их итоги."
        print(f"Выполнение запроса: {message}\\\\n")

        # Генерируйте идентификатор трассировки для отладки
        trace_id = gen_trace_id()
        with trace(workflow_name="Пример файловой системы MCP", trace_id=trace_id):
            print(f"Просмотр трассировки: <https://platform.openai.com/traces/{trace_id}\\\\n>")
            result = await Runner.run(starting_agent=agent, input=message)
            print(result.final_output)

if __name__ == "__main__":
    # Проверьте, установлен ли npx
    if not shutil.which("npx"):
        raise RuntimeError("npx не установлен. Пожалуйста, установите его с помощью `npm install -g npx`.")

    asyncio.run(run_agent_with_mcp())

В этом примере:

1. Мы создаем экземпляр MCPServerStdio, который запускает сервер файловой системы MCP как подпроцесс.
2. Мы передаем этот сервер конструктору Agent через параметр mcp_servers.
3. Когда агент работает, он автоматически вызывает list_tools() на MCP сервере, чтобы сделать LLM осведомленной о доступных инструментах.
4. Если LLM решает использовать любой из инструментов MCP, SDK вызывает call_tool() на сервере.

Подключение к серверу MCP SSE

Теперь давайте посмотрим, как подключиться к удаленному серверу MCP с использованием SSE:

Шаг 1: Понимание серверов MCP SSE

Серверы MCP (Server-Sent Events) SSE работают удаленно и предоставляют свою функциональность через HTTP конечные точки. В отличие от серверов stdio, они не работают как подпроцессы вашего приложения.

Шаг 2: Подключение к серверу MCP SSE

Вот пример кода, подключающегося к серверу MCP SSE:

import asyncio
from agents import Agent, Runner, gen_trace_id, trace
from agents.mcp import MCPServerSse
from agents.model_settings import ModelSettings

async def run_agent_with_remote_mcp():
    # Создайте и настройте соединение с сервером SSE MCP
    async with MCPServerSse(
        name="Служба погоды",
        params={
            "url": "<https://example.com/mcp/sse>",
            # Необязательные параметры аутентификации
            "headers": {
                "Authorization": "Bearer ваш_ключ_api_здесь"
            }
        },
    ) as mcp_server:
        # Создайте агента, использующего удаленный сервер MCP
        agent = Agent(
            name="WeatherAssistant",
            instructions="Используйте инструменты погоды, чтобы ответить на вопросы о текущих погодных условиях.",
            mcp_servers=[mcp_server],
            # Принудите агента использовать инструменты, когда они доступны
            model_settings=ModelSettings(tool_choice="required")
        )

        # Запустите агента с запросом от пользователя
        message = "Какая погода в Токио сегодня?"
        print(f"Выполнение запроса: {message}\\\\n")

        trace_id = gen_trace_id()
        with trace(workflow_name="Пример погоды MCP", trace_id=trace_id):
            print(f"Просмотр трассировки: <https://platform.openai.com/traces/{trace_id}\\\\n>")
            result = await Runner.run(starting_agent=agent, input=message)
            print(result.final_output)

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

Создание простого локального сервера MCP SSE

Чтобы полностью понять, как работает MCP, полезно реализовать простой MCP сервер. Вот пример минимального сервера MCP SSE:

import asyncio
import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from typing import Dict, Any, List, Optional

app = FastAPI()

# Определите инструменты, которые будет предоставлять наш MCP сервер
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "add",
            "description": "Сложите два числа",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "number", "description": "Первое число"},
                    "b": {"type": "number", "description": "Второе число"}
                },
                "required": ["a", "b"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Получить погоду для местоположения",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "Название города"}
                },
                "required": ["location"]
            }
        }
    }
]

# Реализуйте фактическую функциональность инструмента
async def call_tool(tool_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
    if tool_name == "add":
        return {"result": parameters["a"] + parameters["b"]}

    elif tool_name == "get_weather":
        # В реальной реализации вы бы вызвали реальный API погоды
        weather_data = {
            "Tokyo": {"condition": "Солнечно", "temperature": 25},
            "New York": {"condition": "Облачно", "temperature": 18},
            "London": {"condition": "Дождливо", "temperature": 15}
        }
        location = parameters["location"]
        if location in weather_data:
            return {"weather": weather_data[location]}
        return {"error": f"Данные о погоде недоступны для {location}"}

    return {"error": f"Неизвестный инструмент: {tool_name}"}

async def sse_event_generator(request: Request):
    # Чтение тела запроса
    body_bytes = await request.body()
    body = json.loads(body_bytes)

    # Обработка различных операций MCP
    if body["action"] == "list_tools":
        yield f"data: {json.dumps({'tools': TOOLS})}\\\\n\\\\n"

    elif body["action"] == "call_tool":
        tool_name = body["tool_name"]
        parameters = body["parameters"]
        result = await call_tool(tool_name, parameters)
        yield f"data: {json.dumps({'result': result})}\\\\n\\\\n"

@app.post("/sse")
async def sse_endpoint(request: Request):
    return StreamingResponse(
        sse_event_generator(request),
        media_type="text/event-stream"
    )

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

Расширенные функции

Кэширование списков инструментов

Для повышения производительности вы можете кэшировать список инструментов с серверов MCP:

# Создайте сервер MCP с кэшированием инструментов
async with MCPServerSse(
    name="Служба погоды",
    params={"url": "<https://example.com/mcp/sse>"},
    cache_tools_list=True  # Включите кэширование
) as mcp_server:
    # Используйте сервер как раньше...

Когда cache_tools_list=True установлен, SDK будет вызывать list_tools() только один раз на сервере MCP и повторно использовать результат для последующих запусков агента. Это снижает задержку, особенно для удаленных серверов.

Чтобы отменить кэш, если это необходимо:

mcp_server.invalidate_tools_cache()

Трассировка операций MCP

SDK агентов OpenAI включает в себя встроенные возможности трассировки, которые автоматически фиксируют операции MCP:

1. Вызовы к серверу MCP для перечисления инструментов
2. Информация, относящаяся к MCP, о вызовах функций

Вы можете просмотреть эти трассировки на https://platform.openai.com/traces/, когда вы используете контекстный менеджер trace, как показано в приведенных выше примерах.

Использование нескольких серверов MCP

Вы можете одновременно подключать вашего агента к нескольким серверам MCP, предоставляя ему доступ к более широкому спектру инструментов:

async def run_with_multiple_servers():
    async with MCPServerStdio(
        name="Файловая система",
        params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "./data"]}
    ) as fs_server, MCPServerSse(
        name="Weather API",
        params={"url": "<https://example.com/weather/mcp/sse>"}
    ) as weather_server:
        # Создайте агента с обоими серверами MCP
        agent = Agent(
            name="MultiToolAssistant",
            instructions="Используйте все доступные инструменты, чтобы помочь пользователю.",
            mcp_servers=[fs_server, weather_server]
        )

        # Запустите агента
        result = await Runner.run(
            starting_agent=agent,
            input="Сначала проверьте погоду в Токио, затем прочитайте содержимое файла report.txt."
        )
        print(result.final_output)

Обработка ошибок и отладка

При работе с серверами MCP вы можете столкнуться с различными проблемами. Вот некоторые распространенные проблемы и как с ними справляться:

Проблемы с подключением

Если ваш сервер MCP не отвечает:

try:
    async with MCPServerSse(
        name="Служба погоды",
        params={"url": "<https://example.com/mcp/sse>"}
    ) as mcp_server:
        # Используйте сервер...
except Exception as e:
    print(f"Не удалось подключиться к серверу MCP: {e}")
    # Реализуйте стратегию резервного копирования

Ошибки выполнения инструментов

Когда выполнение инструмента терпит неудачу, обрабатывайте это аккуратно:

try:
    result = await Runner.run(starting_agent=agent, input=user_query)
    print(result.final_output)
except Exception as e:
    print(f"Ошибка во время выполнения агента: {e}")
    # Возможно, вам стоит проверить журналы трассировки для получения подробной информации об ошибке

Заключение

Поддержка MCP в SDK агентов OpenAI позволяет расширять ваших агентов с помощью широкого спектра инструментов и возможностей. Независимо от того, используете ли вы локальные серверы stdio или удаленные конечные точки SSE, интеграция проста и мощна.

Подключив к серверам MCP, ваши агенты могут получить доступ к файловым системам, API погоды, базам данных и практически любому внешнему инструменту или источнику данных, который открывает интерфейс MCP. Эта гибкость делает SDK агента OpenAI мощной основой для создания сложных AI-приложений.

Не забудьте использовать такие функции, как кэширование инструментов, чтобы оптимизировать производительность, и используйте встроенные возможности трассировки для отладки и мониторинга взаимодействий вашего агента с серверами MCP.

Удачного программирования с OpenAI Agents и MCP!