Интеграция с LLM

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

#1. Обзор интеграций

MCP серверы могут работать с различными LLM через MCP Host:

LLM	Способ подключения	Сложность
Claude (Anthropic)	Claude Desktop	Низкая
GPT (OpenAI)	Через MCP-совместимые клиенты	Средняя
Ollama (локальные)	Через MCP-клиенты	Средняя
LM Studio	Через MCP-плагин	Низкая

#2. Claude Desktop (официальный способ)

#Настройка конфигурации

macOS: ~/Library/Application Support/Claude/claude_desktop_config.json

Windows: %APPDATA%\Claude\claude_desktop_config.json

Linux: ~/.config/Claude/claude_desktop_config.json

#Конфигурация для stdio сервера

{
  "mcpServers": {
    "my-local-server": {
      "command": "poetry",
      "args": ["run", "python", "src/server.py"],
      "cwd": "/path/to/project",
      "env": {
        "API_KEY": "your-secret-key",
        "DEBUG": "true"
      }
    },
    "another-server": {
      "command": "node",
      "args": ["/path/to/server.js"],
      "env": {}
    }
  }
}

#Конфигурация для SSE сервера

{
  "mcpServers": {
    "remote-sse-server": {
      "url": "http://localhost:8000/sse"
    },
    "cloud-server": {
      "url": "https://mcp.example.com/sse",
      "headers": {
        "Authorization": "Bearer your-token"
      }
    }
  }
}

#Проверка подключения

Откройте Claude Desktop
Перейдите в Settings → MCP
Убедитесь, что серверы в списке со статусом "Connected"
Если статус "Error", проверьте логи

#Использование в диалоге

После подключения Claude автоматически видит доступные инструменты:

Пользователь: Покажи погоду в Москве

Claude: [Использует инструмент get_weather с city="Moscow"]
        Сейчас в Москве +20°C, солнечно...

#3. Интеграция с GPT (OpenAI)

#Через MCP-совместимые клиенты

OpenAI официально не поддерживает MCP, но можно использовать совместимые клиенты:

#Вариант 1: Через сторонние MCP Host

# Пример использования через mcp-python клиент
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def use_mcp_with_gpt():
    async with stdio_client(
        StdioServerParameters(
            command="poetry",
            args=["run", "python", "src/server.py"]
        )
    ) as (read, write):
        async with ClientSession(read, write) as session:
            # Инициализация
            await session.initialize()
            
            # Получение списка инструментов
            tools = await session.list_tools()
            
            # Вызов инструмента
            result = await session.call_tool("search", {"query": "MCP"})
            
            # Использование результата с GPT
            from openai import OpenAI
            
            client = OpenAI(api_key="sk-...")
            
            response = client.chat.completions.create(
                model="gpt-4",
                messages=[
                    {"role": "system", "content": "You are a helpful assistant."},
                    {"role": "user", "content": f"Search results: {result}"},
                ],
                tools=[  # Function calling
                    {
                        "type": "function",
                        "function": {
                            "name": "search",
                            "description": "Search knowledge base",
                            "parameters": {
                                "type": "object",
                                "properties": {
                                    "query": {"type": "string"}
                                },
                                "required": ["query"]
                            }
                        }
                    }
                ]
            )

#Вариант 2: Адаптер MCP → OpenAI Functions

from openai import OpenAI
from mcp import ClientSession
import json


class MCPToOpenAIAdapter:
    """Адаптирует MCP инструменты для OpenAI Function Calling."""
    
    def __init__(self, mcp_session: ClientSession):
        self.session = mcp_session
        self.openai_client = OpenAI(api_key="sk-...")
    
    async def get_openai_tools(self):
        """Получить MCP инструменты в формате OpenAI."""
        mcp_tools = await self.session.list_tools()
        
        return [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema
                }
            }
            for tool in mcp_tools.tools
        ]
    
    async def execute_mcp_tool(self, name: str, arguments: dict):
        """Выполнить MCP инструмент."""
        result = await self.session.call_tool(name, arguments)
        return json.dumps(result.content)
    
    async def chat(self, user_message: str):
        """Чат с использованием MCP инструментов."""
        # Получаем инструменты
        tools = await self.get_openai_tools()
        
        # Первый запрос к GPT
        response = self.openai_client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "Use tools when helpful."},
                {"role": "user", "content": user_message}
            ],
            tools=tools
        )
        
        # Проверяем, хочет ли GPT вызвать инструмент
        message = response.choices[0].message
        
        if message.tool_calls:
            # Выполняем MCP инструмент
            for tool_call in message.tool_calls:
                result = await self.execute_mcp_tool(
                    tool_call.function.name,
                    json.loads(tool_call.function.arguments)
                )
                
                # Отправляем результат обратно в GPT
                response = self.openai_client.chat.completions.create(
                    model="gpt-4",
                    messages=[
                        {"role": "system", "content": "Use tools when helpful."},
                        {"role": "user", "content": user_message},
                        message,
                        {
                            "role": "tool",
                            "tool_call_id": tool_call.id,
                            "content": result
                        }
                    ]
                )
        
        return response.choices[0].message.content

#4. Локальные модели (Ollama)

#Настройка Ollama

# Установка Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# Загрузка модели
ollama pull llama3.1

# Запуск сервера
ollama serve

#Интеграция через MCP-клиент

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import requests
import json


class OllamaMCPIntegration:
    """Интеграция Ollama с MCP сервером."""
    
    def __init__(self, mcp_session: ClientSession, ollama_url: str = "http://localhost:11434"):
        self.session = mcp_session
        self.ollama_url = ollama_url
        self.model = "llama3.1"
    
    def generate(self, prompt: str, tools_context: str = "") -> str:
        """Генерация ответа через Ollama."""
        full_prompt = prompt
        if tools_context:
            full_prompt = f"""
You have access to the following tools:
{tools_context}

User request: {prompt}
"""
        
        response = requests.post(
            f"{self.ollama_url}/api/generate",
            json={
                "model": self.model,
                "prompt": full_prompt,
                "stream": False
            }
        )
        
        return response.json()["response"]
    
    async def chat_with_tools(self, user_message: str):
        """Чат с использованием MCP инструментов."""
        # Получаем инструменты
        tools = await self.session.list_tools()
        
        # Формируем контекст инструментов
        tools_context = "\n".join([
            f"- {tool.name}: {tool.description}\n"
            f"  Parameters: {json.dumps(tool.inputSchema)}"
            for tool in tools.tools
        ])
        
        # Первый запрос
        response = self.generate(user_message, tools_context)
        
        # Простая эвристика для вызова инструментов
        # В реальности нужен более сложный парсинг
        if "call_tool" in response.lower():
            # Парсим вызов инструмента и выполняем
            tool_name = self._parse_tool_name(response)
            tool_args = self._parse_tool_args(response)
            
            result = await self.session.call_tool(tool_name, tool_args)
            
            # Второй запрос с результатом
            final_response = self.generate(
                f"{user_message}\n\nTool result: {json.dumps(result)}",
                tools_context
            )
            return final_response
        
        return response
    
    def _parse_tool_name(self, text: str) -> str:
        # Простая реализация парсинга
        return "search"
    
    def _parse_tool_args(self, text: str) -> dict:
        # Простая реализация парсинга
        return {"query": "test"}


# Использование
async def main():
    async with stdio_client(
        StdioServerParameters(
            command="poetry",
            args=["run", "python", "src/server.py"]
        )
    ) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            integration = OllamaMCPIntegration(session)
            response = await integration.chat_with_tools(
                "Найди информацию о MCP протоколе"
            )
            print(response)

#5. LM Studio Integration

#Настройка LM Studio

Скачайте LM Studio с lmstudio.ai
Загрузите модель (например, Llama 3.1)
Включите MCP в настройках

#Конфигурация MCP в LM Studio

{
  "mcp": {
    "enabled": true,
    "servers": {
      "local-server": {
        "command": "poetry",
        "args": ["run", "python", "src/server.py"],
        "cwd": "/path/to/project"
      }
    }
  }
}

#Использование

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

#6. Кастомный MCP Host

#Создание собственного хоста

#!/usr/bin/env python3
"""Пример кастомного MCP Host с интеграцией LLM."""

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from openai import OpenAI
import asyncio
import json


class CustomMCPHost:
    """Кастомный MCP Host с интеграцией OpenAI."""
    
    def __init__(self, openai_api_key: str):
        self.openai_client = OpenAI(api_key=openai_api_key)
        self.session = None
    
    async def connect_to_server(self, command: str, args: list):
        """Подключение к MCP серверу."""
        self.stdio_context = stdio_client(
            StdioServerParameters(command=command, args=args)
        )
        
        read, write = await self.stdio_context.__aenter__()
        self.session = ClientSession(read, write)
        await self.session.initialize()
    
    async def get_available_tools(self):
        """Получить доступные инструменты."""
        tools_response = await self.session.list_tools()
        return tools_response.tools
    
    def convert_to_openai_format(self, mcp_tools):
        """Конвертация MCP инструментов в формат OpenAI."""
        return [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema
                }
            }
            for tool in mcp_tools
        ]
    
    async def process_user_message(self, message: str):
        """Обработка сообщения пользователя."""
        # Получаем инструменты
        mcp_tools = await self.get_available_tools()
        openai_tools = self.convert_to_openai_format(mcp_tools)
        
        # Запрос к GPT
        response = self.openai_client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are a helpful assistant with access to tools."},
                {"role": "user", "content": message}
            ],
            tools=openai_tools,
            tool_choice="auto"
        )
        
        assistant_message = response.choices[0].message
        
        # Проверяем вызовы инструментов
        if assistant_message.tool_calls:
            results = []
            
            for tool_call in assistant_message.tool_calls:
                tool_name = tool_call.function.name
                tool_args = json.loads(tool_call.function.arguments)
                
                print(f"Calling MCP tool: {tool_name}({tool_args})")
                
                # Вызов MCP инструмента
                result = await self.session.call_tool(tool_name, tool_args)
                results.append({
                    "tool_call_id": tool_call.id,
                    "result": result
                })
            
            # Отправляем результаты обратно в GPT
            final_response = self.openai_client.chat.completions.create(
                model="gpt-4",
                messages=[
                    {"role": "system", "content": "You are a helpful assistant."},
                    {"role": "user", "content": message},
                    assistant_message,
                ] + [
                    {
                        "role": "tool",
                        "tool_call_id": r["tool_call_id"],
                        "content": json.dumps(r["result"].content)
                    }
                    for r in results
                ]
            )
            
            return final_response.choices[0].message.content
        
        return assistant_message.content
    
    async def close(self):
        """Закрытие соединения."""
        if self.session:
            await self.session.__aexit__(None, None, None)
        if hasattr(self, 'stdio_context'):
            await self.stdio_context.__aexit__(None, None, None)


# Пример использования
async def main():
    host = CustomMCPHost(openai_api_key="sk-...")
    
    try:
        await host.connect_to_server(
            command="poetry",
            args=["run", "python", "src/server.py"]
        )
        
        response = await host.process_user_message(
            "Найди информацию о Python MCP"
        )
        print(f"Ответ: {response}")
    
    finally:
        await host.close()


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

#7. Мульти-провайдер интеграция

#Поддержка нескольких LLM

from typing import Optional
from openai import OpenAI
import anthropic


class MultiLLMHost:
    """Хост с поддержкой нескольких LLM провайдеров."""
    
    def __init__(self):
        self.openai_client: Optional[OpenAI] = None
        self.anthropic_client: Optional[anthropic.Anthropic] = None
        self.current_provider = "anthropic"
    
    def configure_openai(self, api_key: str):
        """Настройка OpenAI."""
        self.openai_client = OpenAI(api_key=api_key)
    
    def configure_anthropic(self, api_key: str):
        """Настройка Anthropic."""
        self.anthropic_client = anthropic.Anthropic(api_key=api_key)
    
    def set_provider(self, provider: str):
        """Выбор текущего провайдера."""
        if provider not in ["openai", "anthropic"]:
            raise ValueError(f"Unknown provider: {provider}")
        self.current_provider = provider
    
    def generate(self, prompt: str, tools: list, tool_results: list = None):
        """Генерация ответа через текущего провайдера."""
        if self.current_provider == "openai":
            return self._generate_openai(prompt, tools, tool_results)
        else:
            return self._generate_anthropic(prompt, tools, tool_results)
    
    def _generate_openai(self, prompt: str, tools: list, tool_results: list = None):
        """Генерация через OpenAI."""
        messages = [
            {"role": "system", "content": "You are helpful."},
            {"role": "user", "content": prompt}
        ]
        
        if tool_results:
            for result in tool_results:
                messages.append({
                    "role": "tool",
                    "tool_call_id": result["id"],
                    "content": result["content"]
                })
        
        response = self.openai_client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            tools=tools
        )
        
        return response.choices[0].message
    
    def _generate_anthropic(self, prompt: str, tools: list, tool_results: list = None):
        """Генерация через Anthropic."""
        response = self.anthropic_client.messages.create(
            model="claude-3-sonnet-20240229",
            max_tokens=1024,
            tools=tools,
            messages=[
                {"role": "user", "content": prompt}
            ]
        )
        
        return response.content[0].text

#8. Best Practices интеграции

#1. Обработка ошибок

async def safe_tool_call(session: ClientSession, tool_name: str, args: dict):
    """Безопасный вызов инструмента с обработкой ошибок."""
    try:
        result = await session.call_tool(tool_name, args)
        return {"success": True, "result": result}
    except Exception as e:
        return {
            "success": False,
            "error": str(e),
            "tool": tool_name
        }

#2. Логирование

import logging

logger = logging.getLogger(__name__)


async def logged_tool_call(session: ClientSession, tool_name: str, args: dict):
    """Вызов инструмента с логированием."""
    logger.info(f"Calling tool: {tool_name} with args: {args}")
    
    start_time = asyncio.get_event_loop().time()
    result = await session.call_tool(tool_name, args)
    duration = asyncio.get_event_loop().time() - start_time
    
    logger.info(f"Tool {tool_name} completed in {duration:.2f}s")
    
    return result

#3. Rate Limiting

from aiolimiter import AsyncLimiter


class RateLimitedMCPHost:
    """MCP Host с ограничением частоты запросов."""
    
    def __init__(self, session: ClientSession, requests_per_minute: int = 60):
        self.session = session
        self.limiter = AsyncLimiter(requests_per_minute, 60)
    
    async def call_tool(self, tool_name: str, args: dict):
        """Вызов инструмента с rate limiting."""
        async with self.limiter:
            return await self.session.call_tool(tool_name, args)

#Ключевые выводы

Провайдер	Способ интеграции	Сложность
Claude	Claude Desktop (официально)	Низкая
GPT	Через адаптер MCP → OpenAI Functions	Средняя
Ollama	Кастомная интеграция	Средняя
LM Studio	Встроенная поддержка MCP	Низкая

Следующая тема: Продвинутые паттерны — кэширование, пулы соединений, асинхронность.

LLM

Способ подключения

Сложность

Claude (Anthropic)

Claude Desktop

Низкая

GPT (OpenAI)

Через MCP-совместимые клиенты

Средняя

Ollama (локальные)

Через MCP-клиенты

Средняя

LM Studio

Через MCP-плагин

Низкая

Пользователь: Покажи погоду в Москве Claude: [Использует инструмент get_weather с city="Moscow"] Сейчас в Москве +20°C, солнечно...

Провайдер

Способ интеграции

Сложность

Claude

Claude Desktop (официально)

Низкая

GPT

Через адаптер MCP → OpenAI Functions

Средняя

Ollama

Кастомная интеграция

Средняя

LM Studio

Встроенная поддержка MCP

Низкая