Работа GigaChat-агента с MCP-сервером

Обновлено 15 апреля 2025

Пример демонстрирует работу агента с MCP-сервером (math_server.py), который дает доступ к функциям сложения, умножения и поиска данных человека по имени. Доступны два варианта агента:

• agent.py — реализует MCP-клиент для локального взаимодействия и не требует отдельного запуска MCP-сервера;
• agent_http.py — реализует MCP-клиент для взаимодействия по HTTP. Для работы агента нужно запустить MCP-серввер в режиме SSE.

Исходный код этого и других примеров вы найдете в GitHub-репозитории.

Агент разработан с помощью фреймворков LangChain и LangGraph, а также библиотеки langchain-mcp-adapters. Библиотека предоставляет обертку для функций, описанных в формате MCP, и позволяет использовать их совместно с LangChain и LangGraph.

В роли LLM агент использует модель GigaChat. Обмен сообщениями с моделью выполняется через GigaChat API.

Model Context Protocol (MCP)

Model Context Protocol (или MCP) — открытый протокол, который унифицирует обмен контекстом между приложением и LLM. Другими словами, использование MCP упрощает подключение больших языковых моделей к различным функциям (инструментам) и источникам данных.

Протокол предоставляет:

• растущий список готовых интеграций, доступных для подключения LLM;
• простоту переключения между различными моделями и их поставщиками;
• набор лучших практик для обеспечения безопасности данных внутри вашей инфраструктуры.

MCP разработан компанией Anthropic. Подробнее о протоколе — в официальной документации.

Подготовка к работе

Чтобы начать работу с примером:

1. Склонируйте репозиторий GigaChain.

2. Перейдите в папку примера:

   cd cookbook/mcp/

3. Установите зависимости:

   pip install langchain-gigachat langchain_mcp_adapters langgraph rich

4. В папке примера создайте файл с переменными окружения .env и добавьте в него переменную GIGACHAT_CREDENTIALS:

GIGACHAT_CREDENTIALS=<ключ_авторизации>

О том как получить ключ авторизации — в разделе Быстрый старт.

Вы также можете указать другие переменные окружения, которые поддерживает Python-библиотека GigaChat.

Быстрый старт

Пример содержит два варианта агента: с локальным и с HTTP-клиентом сервера MCP.

Исходный код MCP-сервера

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel
import sys

mcp = FastMCP("Math")

class Query(BaseModel):
    query: str


class Person(BaseModel):
    name: str
    age: int


@mcp.tool()
def find_preson(name: Query) -> Person:
    """Find an info about some person by name"""
    print(f">>> Calling find_person({name})")
    return Person(name="John Doe", age=30)


@mcp.tool()
def add(a: float, b: float) -> float:
    """Add two numbers"""
    print(f">>> Calling add({a}, {b})")
    return a + b


@mcp.tool()
def multiply(a: float, b: float) -> float:
    """Multiply two numbers"""
    print(f">>> Calling multiply({a}, {b})")
    return a * b


if __name__ == "__main__":
    transport = sys.argv[1] if len(sys.argv) > 1 else "stdio"
    mcp.run(transport=transport)

• Локальный клиент
• HTTP-клиент

Для запуска локального клиента выполните команду:

python agent.py

Исходный код агента с локальным клиентом

import asyncio
import os

from dotenv import find_dotenv, load_dotenv
from langchain_gigachat.chat_models.gigachat import GigaChat
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from rich import print as rprint

load_dotenv(find_dotenv())

# Инициализация GigaChat
model = GigaChat(model="GigaChat-2-Max",
                streaming=False,
                max_tokens=8000,
                timeout=600)

def _log(ans):
    for message in ans['messages']:
        rprint(f"[{type(message).__name__}] {message.content} {getattr(message, 'tool_calls', '')}")


async def main():
    server_params = StdioServerParameters(
        command="python",
        args=["math_server.py"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Инициализация соединения с сервером
            await session.initialize()

            # Загрузка функций с сервера
            tools = await load_mcp_tools(session)

            # Создание и запуск агента
            agent = create_react_agent(model, tools)

            agent_response = await agent.ainvoke({"messages": [
                {"role": "user", "content": "Сколько будет (3 + 5) x 12?"}]})
            _log(agent_response)
            
            agent_response = await agent.ainvoke({"messages": [
                {"role": "user", "content": "Найди сколько лет Джону Доу?"}]})
            _log(agent_response)

# Запуск функции main
asyncio.run(main())

Для запуска HTTP-клиента:

1. Запустите MCP-сервер в режиме SSE с помощью команды:

   python math_server.py sse

2. Запустите клиент с помощью команды:

   python agent_http.py

Исходный код агента с HTTP-клиентом

import asyncio
import os

from dotenv import find_dotenv, load_dotenv
from langchain_gigachat.chat_models.gigachat import GigaChat
from langgraph.prebuilt import create_react_agent
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.client import MultiServerMCPClient
from rich import print as rprint

load_dotenv(find_dotenv())

# Инициализация GigaChat
model = GigaChat(model="GigaChat-2-Max",
                streaming=False,
                max_tokens=8000,
                timeout=600)


def _log(ans):
    for message in ans['messages']:
        rprint(f"[{type(message).__name__}] {message.content} {getattr(message, 'tool_calls', '')}")


async def main():
    async with MultiServerMCPClient(
        {
            "math": {
                "url": "http://localhost:8000/sse",
                "transport": "sse",
            }
        }
    ) as client:
        agent = create_react_agent(model, client.get_tools())
        
        agent_response = await agent.ainvoke({"messages": [
            {"role": "user", "content": "Сколько будет (3 + 5) x 12?"}]})
        _log(agent_response)
        
        agent_response = await agent.ainvoke({"messages": [
            {"role": "user", "content": "Найди сколько лет Джону Доу?"}]})
        _log(agent_response)

# Запуск функции main
asyncio.run(main())

Пример вывода агента:

[HumanMessage] Сколько будет (3 + 5) x 12? 
[AIMessage]  [{'name': 'add', 'args': {'a': 3, 'b': 5}, 'id': '99f7f6c7-baac-4e61-9577-03903e83f3a7', 'type': 'tool_call'}]
[ToolMessage] 8.0 
[AIMessage]  [{'name': 'multiply', 'args': {'a': 8, 'b': 12}, 'id': 'c923315e-0888-47c3-a380-2f91d95c3177', 'type': 'tool_call'}]
[ToolMessage] 96.0 
[AIMessage] Результат выражения $(3+5)\times12$ равен $96$. []
[HumanMessage] Найди сколько лет Джону Доу? 
[AIMessage]  [{'name': 'find_preson', 'args': {'name': {'query': 'Джон Доу'}}, 'id': 'fa2ecddc-c446-477b-adc7-7d4f09281953', 'type': 'tool_call'}]
[ToolMessage] {"name": "John Doe", "age": 30} 
[AIMessage] Джону Доу 30 лет. []