Привет, чемпионы!
Казалось бы, совсем недавно мир только начал знакомиться с тем, что такое большие языковые модели (LLM). Вскоре после этого появились их многочисленные вариации - на любой вкус и цвет, от узкоспециализированных до универсальных моделей. Затем началась волна интеграций: LLM начали встраивать в различные сервисы, приложения и API, упрощая и автоматизируя рутинные процессы.

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

И вот, компания Anthropic представила решение этой задачи - новый протокол Model Context Protocol (MCP), который стандартизирует взаимодействие агентов с различными сервисами и между собой.

Давайте разберёмся, что такое MCP, и с чем его едят!

MCP или объясняем на пальцах взаимодействие LLM с внешним миром

Ещё недавно каждый AI-агент был как самостоятельный турист: чтобы заказать билет, найти отель или трансфер, ему приходилось настраивать отдельный способ общения с каждым сервисом. Один API требовал одного формата, другой - совершенно другое. В результате - множество интеграций, лишняя сложность и постоянные доработки.

Представьте теперь, что появился единый язык общения - MCP. Агент больше не тратит время на индивидуальные настройки для каждого сервиса. Он отправляет стандартный запрос, который сразу понятен всем сервисам без дополнительной адаптации.

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

Основные возможности и особенности протокола

Протокол MCP часто сравнивают с USB-C за его способность унифицировать взаимодействие с различными внешними инструментами и источниками информации.

Сервер MCP помогает легко находить доступные инструменты и сразу понимать, как с ними работать и что изменилось.

Представьте, вы создали API, который отправляет фотографии и текст за последние сутки. Затем решили расширить функционал и добавить аудио. Без обновления клиентского кода неизбежны ошибки и потери данных.

Но с MCP всё гораздо проще:

• Модель отправляет запрос на сервер, используя стандарт MCP.

• Вместо жёстко заданной схемы сервер динамически описывает свои доступные данные через обновлённый контекст.

• При следующем обмене модель автоматически узнаёт, что теперь доступны не только изображения и текст, но и аудиофайлы.

• Тем самым LLM может без ошибок адаптировать своё поведение под новые возможности - без изменения кода вручную.

Теперь давайте обсудим основные ключевые моменты и перейдем к краткому примеру работы.

Архитектура MCP состоит из трёх главных компонентов:

• Хост (Host): Приложение, в котором работает LLM (например, Claude Desktop, ваша IDE или Docker) .

• Клиент (Client): Компонент, встроенный в хост, который устанавливает и управляет соединениями с MCP-серверами.

• Сервер (Server): Отдельный процесс, предоставляющий определенные ресурсы, инструменты или данные, доступные через стандартизированный протокол MCP.​

Каждый клиент поддерживает индивидуальное соединение с сервером, обеспечивая изоляцию и безопасность взаимодействия.

Серверы MCP предоставляют три типа ресурсов, называемых примитивами:

• Инструменты (Tools): Функции, которые LLM может вызывать для выполнения определенных действий (например, API-запросы или работа с файлами).

• Ресурсы (Resources): Контекстные данные, предоставляемые сервером, например, содержимое файлов или результаты запросов.

• Подсказки (Prompts): Шаблоны или инструкции, которые могут быть использованы для формирования запросов к LLM.

Гифка ниже прекрасно иллюстрирует эту концепцию!

Пример работы MCP

Сделаем простую реализацию из mcp серверов, клиента и LLM, в качестве LLM возьмем свеженькую qwen3:1.7b.

Архитектура будет примерно следующая:

• LLM: Qwen3 на 1.7b параметров.

• Клиент MCP: на основе библиотеки langchain_mcp_adapters.

• Сервера MCP: первый - для поиска в интернете с помощью duckduckgo, второй - простой сервер для взаимодействия с файлами.

from mcp.server.fastmcp import FastMCP, Context
import httpx
from bs4 import BeautifulSoup
from typing import List, Dict, Optional, Any
from dataclasses import dataclass
import urllib.parse
import sys
import traceback
import asyncio
from datetime import datetime, timedelta
import time
import re


@dataclass
class SearchResult:
    title: str
    link: str
    snippet: str
    position: int


class RateLimiter:
    def __init__(self, requests_per_minute: int = 30):
        self.requests_per_minute = requests_per_minute
        self.requests = []

    async def acquire(self):
        now = datetime.now()
        # Remove requests older than 1 minute
        self.requests = [
            req for req in self.requests if now - req < timedelta(minutes=1)
        ]

        if len(self.requests) >= self.requests_per_minute:
            # Wait until we can make another request
            wait_time = 60 - (now - self.requests[0]).total_seconds()
            if wait_time > 0:
                await asyncio.sleep(wait_time)

        self.requests.append(now)


class DuckDuckGoSearcher:
    BASE_URL = "https://html.duckduckgo.com/html"
    HEADERS = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
    }

    def __init__(self):
        self.rate_limiter = RateLimiter()

    def format_results_for_llm(self, results: List[SearchResult]) -> str:
        """Format results in a natural language style that's easier for LLMs to process"""
        if not results:
            return "No results were found for your search query. This could be due to DuckDuckGo's bot detection or the query returned no matches. Please try rephrasing your search or try again in a few minutes."

        output = []
        output.append(f"Found {len(results)} search results:\n")

        for result in results:
            output.append(f"{result.position}. {result.title}")
            output.append(f"   URL: {result.link}")
            output.append(f"   Summary: {result.snippet}")
            output.append("")  # Empty line between results

        return "\n".join(output)

    async def search(
        self, query: str, ctx: Context, max_results: int = 10
    ) -> List[SearchResult]:
        try:
            await self.rate_limiter.acquire()

            data = {
                "q": query,
                "b": "",
                "kl": "",
            }

            await ctx.info(f"Searching DuckDuckGo for: {query}")

            async with httpx.AsyncClient() as client:
                response = await client.post(
                    self.BASE_URL, data=data, headers=self.HEADERS, timeout=30.0
                )
                response.raise_for_status()

            soup = BeautifulSoup(response.text, "html.parser")
            if not soup:
                await ctx.error("Failed to parse HTML response")
                return []

            results = []
            for result in soup.select(".result"):
                title_elem = result.select_one(".result__title")
                if not title_elem:
                    continue

                link_elem = title_elem.find("a")
                if not link_elem:
                    continue

                title = link_elem.get_text(strip=True)
                link = link_elem.get("href", "")

                # Skip ad results
                if "y.js" in link:
                    continue

                # Clean up DuckDuckGo redirect URLs
                if link.startswith("//duckduckgo.com/l/?uddg="):
                    link = urllib.parse.unquote(link.split("uddg=")[1].split("&")[0])

                snippet_elem = result.select_one(".result__snippet")
                snippet = snippet_elem.get_text(strip=True) if snippet_elem else ""

                results.append(
                    SearchResult(
                        title=title,
                        link=link,
                        snippet=snippet,
                        position=len(results) + 1,
                    )
                )

                if len(results) >= max_results:
                    break

            await ctx.info(f"Successfully found {len(results)} results")
            return results

        except httpx.TimeoutException:
            await ctx.error("Search request timed out")
            return []
        except httpx.HTTPError as e:
            await ctx.error(f"HTTP error occurred: {str(e)}")
            return []
        except Exception as e:
            await ctx.error(f"Unexpected error during search: {str(e)}")
            traceback.print_exc(file=sys.stderr)
            return []


class WebContentFetcher:
    def __init__(self):
        self.rate_limiter = RateLimiter(requests_per_minute=20)

    async def fetch_and_parse(self, url: str, ctx: Context) -> str:
        """Fetch and parse content from a webpage"""
        try:
            await self.rate_limiter.acquire()

            await ctx.info(f"Fetching content from: {url}")

            async with httpx.AsyncClient() as client:
                response = await client.get(
                    url,
                    headers={
                        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
                    },
                    follow_redirects=True,
                    timeout=30.0,
                )
                response.raise_for_status()

            # Parse the HTML
            soup = BeautifulSoup(response.text, "html.parser")

            # Remove script and style elements
            for element in soup(["script", "style", "nav", "header", "footer"]):
                element.decompose()

            # Get the text content
            text = soup.get_text()

            # Clean up the text
            lines = (line.strip() for line in text.splitlines())
            chunks = (phrase.strip() for line in lines for phrase in line.split("  "))
            text = " ".join(chunk for chunk in chunks if chunk)

            # Remove extra whitespace
            text = re.sub(r"\s+", " ", text).strip()

            # Truncate if too long
            if len(text) > 8000:
                text = text[:8000] + "... [content truncated]"

            await ctx.info(
                f"Successfully fetched and parsed content ({len(text)} characters)"
            )
            return text

        except httpx.TimeoutException:
            await ctx.error(f"Request timed out for URL: {url}")
            return "Error: The request timed out while trying to fetch the webpage."
        except httpx.HTTPError as e:
            await ctx.error(f"HTTP error occurred while fetching {url}: {str(e)}")
            return f"Error: Could not access the webpage ({str(e)})"
        except Exception as e:
            await ctx.error(f"Error fetching content from {url}: {str(e)}")
            return f"Error: An unexpected error occurred while fetching the webpage ({str(e)})"


# Initialize FastMCP server
mcp = FastMCP("ddg-search")
searcher = DuckDuckGoSearcher()
fetcher = WebContentFetcher()


@mcp.tool()
async def search(query: str, ctx: Context, max_results: int = 10) -> str:
    """
    Search DuckDuckGo and return formatted results.

    Args:
        query: The search query string
        max_results: Maximum number of results to return (default: 10)
        ctx: MCP context for logging
    """
    try:
        results = await searcher.search(query, ctx, max_results)
        return searcher.format_results_for_llm(results)
    except Exception as e:
        traceback.print_exc(file=sys.stderr)
        return f"An error occurred while searching: {str(e)}"


@mcp.tool()
async def fetch_content(url: str, ctx: Context) -> str:
    """
    Fetch and parse content from a webpage URL.

    Args:
        url: The webpage URL to fetch content from
        ctx: MCP context for logging
    """
    return await fetcher.fetch_and_parse(url, ctx)


def main():
    mcp.run()


if __name__ == "__main__":
    main()

Код сервера mcp duckduckgo

import os
from mcp.server.fastmcp import FastMCP
from core.cfg import CFG

mcp = FastMCP("FileTools")

BASE_DIR = CFG.BASE_DIR

os.makedirs(BASE_DIR, exist_ok=True)

@mcp.tool()
def create_folder(folder_name: str) -> str:
    """Создаёт папку с указанным именем в базовой директории."""
    folder_path = os.path.join(BASE_DIR, folder_name)
    try:
        os.makedirs(folder_path, exist_ok=True)
        return f"Папка '{folder_name}' успешно создана."
    except Exception as e:
        return f"Ошибка при создании папки: {e}"

@mcp.tool()
def create_text_file(file_name: str, content: str) -> str:
    """Создаёт текстовый файл с указанным именем и содержимым в базовой директории."""
    file_path = os.path.join(BASE_DIR, file_name)
    try:
        with open(file_path, 'w', encoding='utf-8') as file:
            file.write(content)
        return f"Файл '{file_name}' успешно создан."
    except Exception as e:
        return f"Ошибка при создании файла: {e}"

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

Код сервера mcp для работы с файлами

import asyncio
import os
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_ollama import ChatOllama

model = ChatOllama(model="qwen3:1.7b")

async def main():
    project_root = os.path.dirname(os.path.abspath(__file__))
    
    async with MultiServerMCPClient({
        "search": {
            "command": "python",
            "args": ["./mcp_server/search_sever_duckduck_go.py"],
            "transport": "stdio",
            "env": {"PYTHONPATH": project_root}
        },
        "files": {
            "command": "python",
            "args": ["./mcp_server/server_2.py"],
            "transport": "stdio",
            "env": {"PYTHONPATH": project_root}
        }
    }) as client:
        search_session = client.sessions["search"]
        files_session = client.sessions["files"]

        search_tools = await load_mcp_tools(search_session)
        files_tools = await load_mcp_tools(files_session)

        agent = create_react_agent(model, search_tools + files_tools)

        print("Интерактивный чат запущен. Напишите 'exit' для выхода.\n")

        while True:
            user_input = input("Вы: ").strip()
            if user_input.lower() in {"exit", "quit"}:
                print("Выход из чата.")
                break

            try:
                res = await agent.ainvoke({"messages": user_input})
                for m in res['messages']:
                    print("Агент:", m.content)
            except Exception as e:
                print(f"Ошибка: {e}")

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

Код для работы с mcp серверами с помощью mcp клиента

В данном случае мы создаем клиента mcp, к которому подключаем оба сервера. Далее мы загружаем доступные инструменты, которые будут использоваться агентом с помощью load_mcp_tools и создаем Reasoning and Acting агента с помощью create_react_agent.

Отлично, код вроде написали, но давайте теперь же опробуем на мини примере.

Проверим на простом примере:

find brief information about mcp capabilities and do it in the reports mcp_res.txt folder

После небольшой паузы агент успешно выполняет задачу. Переходим в указанную папку и убеждаемся: всё работает!

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

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

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

? Ставьте лайк и напишите какие темы было бы интересно разобрать дальше! Самое главное - пробуйте и экспериментируйте!

✔️ Присоединяйтесь к нашему Telegram-сообществу @datafeeling, чтобы первыми применять на практике передовые технологии!

https://github.com/Sasha5017/agent_mcp