MCP: инструменты через внешние серверы | Курс LangChain Agents урок 3

Цель урока: подключить агента к серверу MCP, написать свой сервер на FastMCP и заменить заглушки агента-аналитика на реальные инструменты.

Теория

Что такое MCP и зачем он нужен

До появления MCP каждый разработчик интегрировал инструменты вручную: написал агента для Claude, написал свой адаптер для поиска, для базы данных, для браузера. Другой разработчик делал то же самое для GPT-4, третий, для своего фреймворка. Одни и те же инструменты переписывались снова и снова для каждой комбинации агент + инструмент. Это классическая проблема M×N интеграций.

MCP (Model Context Protocol) - открытый протокол, который Anthropic выпустил в ноябре 2024 года. Идея похожа на USB-C: вместо того чтобы каждое устройство использовало свой разъём, появился единый стандарт. Инструмент реализует протокол один раз и его сразу можно подключить к любому агенту, который понимает MCP.

Сегодня вокруг MCP сложилась экосистема: сотни готовых серверов для работы с файлами, браузером, базами данных, GitHub, Slack, почтой и многим другим. Крупные редакторы, IDE и AI-ассистенты поддерживают MCP как стандарт подключения инструментов.

Почему в этом курсе именно MCP, а не только @tool

@tool никуда не делся, это рабочий подход и сотни инструментов из сообщества LangChain (langchain-community и другие) являются обычными функциями с декоратором. Но разработчику агентов в 2026 году важно уметь работать с MCP, потому что:

готовые серверы MCP уже есть для большинства популярных сервисов
инструменты, написанные командой без доступа к вашему коду, приходят через MCP
один сервер работает с любым AI-приложением, а не только с вашим агентом

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

Схема подключения:

Агент → MultiServerMCPClient → сервер MCP 1 (stdio)
                             → сервер MCP 2 (http)
                             → сервер MCP 3 (http)

Транспорты

MCP поддерживает два основных транспорта.

stdio: клиент запускает сервер как подпроцесс и общается через stdin/stdout. Подходит для локальных инструментов: работа с файлами, запросы к базе данных, локальный поиск.

{
    "my_server": {
        "transport": "stdio",
        "command": "python",
        "args": ["/абсолютный/путь/к/server.py"],
    }
}

http: сервер запущен отдельно, клиент обращается по URL. Подходит для удалённых API, сервисов, которые работают независимо. Перед подключением сервер нужно запустить отдельно: python weather_server.py.

{
    "my_server": {
        "transport": "http",
        "url": "http://localhost:8000/mcp",
    }
}

`MultiServerMCPClient`

Это основной класс из пакета langchain-mcp-adapters. Он:

принимает конфигурацию серверов
запрашивает у каждого список инструментов
оборачивает их в объекты, совместимые с LangChain
передаёт агенту

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

Async

Клиент MCP работает асинхронно. Методы get_tools() и ainvoke() требуют await. Весь код с MCP оборачивается в async def main() и запускается через asyncio.run(main()).

Три примитива MCP

Протокол MCP определяет три типа объектов, которые сервер может предоставлять клиенту.

Tools - функции, которые агент вызывает во время работы. Это то же самое, что @tool в LangChain, только находяться на отдельном сервере. Агент решает, когда и с какими аргументами вызвать инструмент.

Resources - данные, которые клиент читает явно до или во время выполнения. Сервер объявляет ресурс с URI вида analyst://templates/report. Клиент запрашивает его через client.get_resources("server_name") и получает объект Blob с содержимым. Это удобно для шаблонов, конфигураций, справочных данных.

Prompts - шаблоны промптов с параметрами. Клиент запрашивает промпт через client.get_prompt("server_name", "prompt_name", arguments={...}) и получает готовые сообщения, которые можно передать агенту. Это позволяет вынести формулировки запросов из кода агента на сервер.

FastMCP

FastMCP это библиотека для создания серверов MCP со всеми тремя примитивами:

from fastmcp import FastMCP

mcp = FastMCP("Название сервера")

@mcp.tool()
def my_tool(param: str) -> str:
    """Описание инструмента."""
    return f"Результат: {param}"

@mcp.resource("myserver://data/config")
def my_config() -> str:
    """Конфигурационные данные."""
    return "key: value"

@mcp.prompt()
def my_prompt(topic: str) -> str:
    """Шаблон промпта."""
    return f"Проанализируй тему: {topic}"

if __name__ == "__main__":
    mcp.run(transport="stdio")

Для запуска через http: mcp.run(transport="streamable-http").

Примеры кода

Установка

pip install langchain-mcp-adapters fastmcp duckduckgo-search

Свой сервер MCP (stdio)

Создайте файл tools_server.py рядом с вашим агентом. Здесь все три примитива: инструменты с реальной реализацией, ресурс с шаблоном отчёта и промпт-шаблон.

Экономия токенов. Реальный поиск возвращает несколько абзацев на каждый результат, контекст быстро растёт. Если вы изучаете механику MCP и хотите сократить расход токенов, замените тело search_web на заглушку: python return f"Результаты поиска по запросу '{query}': тема актуальна, данных достаточно." Для реальных задач оставьте полную реализацию с DuckDuckGo.

from fastmcp import FastMCP
from duckduckgo_search import DDGS
from datetime import datetime

mcp = FastMCP("AnalystTools")


# --- Инструменты ---

@mcp.tool()
def search_web(query: str, max_results: int = 5) -> str:
    """Ищет актуальную информацию в интернете по запросу."""
    try:
        with DDGS() as ddgs:
            results = list(ddgs.text(query, max_results=max_results))
        if not results:
            return f"По запросу '{query}' ничего не найдено."
        lines = []
        for i, r in enumerate(results, 1):
            lines.append(f"{i}. {r['title']}\n   {r['href']}\n   {r['body']}")
        return "\n\n".join(lines)
    except Exception as e:
        return f"Ошибка поиска: {e}"


@mcp.tool()
def read_file(path: str) -> str:
    """Читает содержимое текстового файла используя переданный путь."""
    try:
        with open(path, "r", encoding="utf-8") as f:
            return f.read()
    except FileNotFoundError:
        return f"Файл '{path}' не найден."
    except Exception as e:
        return f"Ошибка: {e}"


@mcp.tool()
def get_current_date() -> str:
    """Возвращает текущую дату и время."""
    return datetime.now().strftime("%Y-%m-%d %H:%M")


# --- Ресурсы ---

@mcp.resource("analyst://templates/report")
def report_template() -> str:
    """Шаблон структуры аналитического отчёта."""
    return """# Аналитический отчёт

**Дата:** {date}
**Тема:** {topic}

## Краткое резюме
[2-3 предложения с главным выводом]

## Ключевые факты
-

## Анализ
[Подробный разбор с источниками]

## Вывод и рекомендации
[Итог и следующие шаги]
"""


# --- Промпты ---

@mcp.prompt()
def analyze_topic(topic: str, depth: str = "подробный") -> str:
    """Промпт-шаблон для запуска анализа темы."""
    return (
        f"Проведи {depth} анализ темы: {topic}\n\n"
        f"1) Найди актуальную информацию через поиск.\n"
        f"2) Структурируй ответ по шаблону аналитического отчёта.\n"
        f"3) Укажи источники для ключевых утверждений."
    )


if __name__ == "__main__":
    mcp.run(transport="stdio")

Агент с сервером MCP

import asyncio
import os
import sys
from pathlib import Path
from dotenv import load_dotenv
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

load_dotenv()

SERVER_PATH = str(Path(__file__).parent / "tools_server.py")

async def main():
    client = MultiServerMCPClient(
        {
            "analyst_tools": {
                "transport": "stdio",
                "command": sys.executable,  # используем тот же Python/venv
                "args": [SERVER_PATH],
            }
        }
    )

    tools = await client.get_tools()
    print(f"Инструменты от MCP: {[t.name for t in tools]}")

    model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))
    agent = create_agent(model=model, tools=tools)

    response = await agent.ainvoke({
        "messages": [{"role": "user", "content": "Найди информацию об Anthropic."}]
    })
    print(response["messages"][-1].content)

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

Обратите внимание: путь к серверу должен быть абсолютным. Используем Path(__file__).parent для надёжности, это работает независимо от того, из какой директории запускается скрипт.

Несколько серверов

client = MultiServerMCPClient(
    {
        "analyst_tools": {
            "transport": "stdio",
            "command": "python",
            "args": ["/путь/к/tools_server.py"],
        },
        "calendar": {
            "transport": "http",
            "url": "http://localhost:8001/mcp",
        },
    }
)

tools = await client.get_tools()
# tools содержит инструменты от обоих серверов

Агент получает объединённый список. Если инструменты имеют одинаковые имена, они будут различаться по префиксу сервера.

Использование ресурсов и промптов

Ресурсы и промпты запрашиваются у сервера отдельно от инструментов, до запуска агента.

Чтение ресурса:

async def main():
    client = MultiServerMCPClient({
        "analyst_tools": {
            "transport": "stdio",
            "command": sys.executable,
            "args": [SERVER_PATH],
        }
    })

    # Получить все ресурсы сервера
    blobs = await client.get_resources("analyst_tools")
    for blob in blobs:
        print(f"URI: {blob.metadata['uri']}")
        print(blob.as_string())

    # Или конкретный ресурс по URI
    blobs = await client.get_resources(
        "analyst_tools",
        uris=["analyst://templates/report"]
    )
    template = blobs[0].as_string()

Клиент возвращает объекты Blob из langchain_core. У каждого есть blob.as_string() для текстового содержимого и blob.metadata["uri"] с адресом ресурса.

Загрузка промпта:

async def main():
    # Получить промпт с аргументами
    messages = await client.get_prompt(
        "analyst_tools",
        "analyze_topic",
        arguments={"topic": "LangChain v1", "depth": "краткий"}
    )
    # messages это список объектов HumanMessage/AIMessage
    for msg in messages:
        print(f"{msg.type}: {msg.content}")

Паттерн: загрузить ресурс как шаблон системного промпта, загрузить промпт как первое сообщение пользователя, затем запустить агента с этими данными.

Сквозной проект: агент-аналитик с MCP

Заменяем заглушки из уроков 1 и 2 на сервер MCP.

import asyncio
import os
import sys
from pathlib import Path
from dotenv import load_dotenv
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

load_dotenv()

# Path(__file__).parent - директория текущего скрипта
# Если tools_server.py лежит рядом: Path(__file__).parent / "tools_server.py"
SERVER_PATH = str(Path(__file__).parent.parent / "agents_course" / "tools_server.py")

SYSTEM_PROMPT = """Ты аналитик-исследователь. Твоя задача - собирать информацию
по запросу пользователя, используя доступные инструменты, и давать структурированный
ответ с выводами.

Если информации недостаточно, сделай несколько поисковых запросов, уточняя каждый
следующий на основе предыдущих результатов."""

async def main():
    client = MultiServerMCPClient(
        {
            "analyst_tools": {
                "transport": "stdio",
                "command": sys.executable,
                "args": [SERVER_PATH],
            }
        }
    )

    tools = await client.get_tools()
    model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o-mini"))

    agent = create_agent(
        model=model,
        tools=tools,
        system_prompt=SYSTEM_PROMPT,
        checkpointer=InMemorySaver(),
    )

    config = {"configurable": {"thread_id": "research-1"}}

    response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "Найди информацию об Anthropic."}]},
        config,
    )
    print(response["messages"][-1].content)

    response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "А кто основатели?"}]},
        config,
    )
    print(response["messages"][-1].content)

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

Частые ошибки

Относительный путь к серверу при stdio

Неправильно:

args = ["tools_server.py"]  # зависит от рабочей директории

Правильно:

args = [str(Path(__file__).parent / "tools_server.py")]  # абсолютный путь

При stdio клиент запускает python tools_server.py как подпроцесс. Если рабочая директория отличается от директории скрипта, файл не будет найден.

invoke вместо ainvoke

# Неправильно: синхронный вызов в async контексте
response = agent.invoke({"messages": [...]})

# Правильно
response = await agent.ainvoke({"messages": [...]})

С MCP агент должен использовать ainvoke. Обычный invoke работает только если нет async зависимостей в инструментах.

Вызов async кода вне asyncio.run

# Неправильно: нельзя await вне async функции
tools = await client.get_tools()

# Правильно: оборачиваем в async def и запускаем
async def main():
    tools = await client.get_tools()
    ...

asyncio.run(main())

Сервер не запустился при stdio

Если сервер MCP падает сразу, агент получит пустой список инструментов или ошибку. Проверьте сервер отдельно:

python tools_server.py

Если вывода нет и процесс завершился без ошибок, значит mcp.run(transport="stdio") ждёт ввода - это нормально. Если есть traceback, исправьте ошибку в сервере.