Подключение MCP к Claude, GPT, локальным моделям. Настройка клиентов.
MCP без LLM — как двигатель без топлива. В этой теме научимся подключать MCP серверы к разным LLM: Claude, GPT, локальным моделям.
MCP серверы могут работать с различными LLM через MCP Host:
| LLM | Способ подключения | Сложность |
|---|---|---|
| Claude (Anthropic) | Claude Desktop | Низкая |
| GPT (OpenAI) | Через MCP-совместимые клиенты | Средняя |
| Ollama (локальные) | Через MCP-клиенты | Средняя |
| LM Studio | Через MCP-плагин | Низкая |
macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
Windows: %APPDATA%\Claude\claude_desktop_config.json
Linux: ~/.config/Claude/claude_desktop_config.json
{
"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": {}
}
}
}{
"mcpServers": {
"remote-sse-server": {
"url": "http://localhost:8000/sse"
},
"cloud-server": {
"url": "https://mcp.example.com/sse",
"headers": {
"Authorization": "Bearer your-token"
}
}
}
}После подключения Claude автоматически видит доступные инструменты:
Пользователь: Покажи погоду в Москве
Claude: [Использует инструмент get_weather с city="Moscow"]
Сейчас в Москве +20°C, солнечно...
OpenAI официально не поддерживает MCP, но можно использовать совместимые клиенты:
# Пример использования через 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"]
}
}
}
]
)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# Установка Ollama
curl -fsSL https://ollama.ai/install.sh | sh
# Загрузка модели
ollama pull llama3.1
# Запуск сервера
ollama servefrom 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){
"mcp": {
"enabled": true,
"servers": {
"local-server": {
"command": "poetry",
"args": ["run", "python", "src/server.py"],
"cwd": "/path/to/project"
}
}
}
}LM Studio автоматически обнаруживает MCP инструменты и предлагает их использовать при генерации.
#!/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())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].textasync 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
}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 resultfrom 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 | Низкая |
Следующая тема: Продвинутые паттерны — кэширование, пулы соединений, асинхронность.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.