Как создать AI-агента LangChain/LangGraph с использованием ClickHouse MCP Server

В этом руководстве вы узнаете, как создать AI-агента LangChain/LangGraph, который может взаимодействовать с SQL-песочницей ClickHouse, используя ClickHouse MCP Server.

Пример ноутбука

Этот пример доступен в виде ноутбука в репозитории примеров.

Предварительные требования

• На вашей системе должен быть установлен Python.
• На вашей системе должен быть установлен pip.
• Вам понадобится ключ API Anthropic или ключ API от другого поставщика LLM.

Вы можете выполнить следующие шаги либо в интерактивной консоли Python (REPL), либо с помощью скрипта.

Установка библиотек

Установите необходимые библиотеки, выполнив следующие команды:

pip install -q --upgrade pip
pip install -q langchain-mcp-adapters langgraph "langchain[anthropic]"

Настройка учетных данных

Далее необходимо указать ваш API-ключ Anthropic:

import os, getpass
os.environ["ANTHROPIC_API_KEY"] = getpass.getpass("Enter Anthropic API Key:")

Введите ключ API Anthropic: ········

Использование другого провайдера LLM

Если у вас нет API-ключа Anthropic и вы хотите использовать другого провайдера LLM, инструкции по настройке учетных данных см. в документации Langchain Providers

Инициализация MCP-сервера

Теперь настройте ClickHouse MCP Server для подключения к песочнице ClickHouse SQL:

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

server_params = StdioServerParameters(
    command="uv",
    args=[
        "run",
        "--with", "mcp-clickhouse",
        "--python", "3.13",
        "mcp-clickhouse"
    ],
    env={
        "CLICKHOUSE_HOST": "sql-clickhouse.clickhouse.com",
        "CLICKHOUSE_PORT": "8443",
        "CLICKHOUSE_USER": "demo",
        "CLICKHOUSE_PASSWORD": "",
        "CLICKHOUSE_SECURE": "true"
    }
)

Настройка обработчика потоков

При работе с Langchain и ClickHouse MCP Server результаты запросов часто возвращаются в виде потоковых данных, а не единым ответом. Для больших наборов данных или сложных аналитических запросов, обработка которых может занять время, важно настроить обработчик потока. Без надлежащей обработки работа с таким потоковым выводом в вашем приложении может быть затруднена.

Настройте обработчик для потокового вывода, чтобы упростить его обработку:

class UltraCleanStreamHandler:
    def __init__(self):
        self.buffer = ""
        self.in_text_generation = False
        self.last_was_tool = False
        
    def handle_chunk(self, chunk):
        event = chunk.get("event", "")
        
        if event == "on_chat_model_stream":
            data = chunk.get("data", {})
            chunk_data = data.get("chunk", {})
            
            # Only handle actual text content, skip tool invocation streams
            if hasattr(chunk_data, 'content'):
                content = chunk_data.content
                if isinstance(content, str) and not content.startswith('{"'):
                    # Add space after tool completion if needed
                    if self.last_was_tool:
                        print(" ", end="", flush=True)
                        self.last_was_tool = False
                    print(content, end="", flush=True)
                    self.in_text_generation = True
                elif isinstance(content, list):
                    for item in content:
                        if (isinstance(item, dict) and 
                            item.get('type') == 'text' and 
                            'partial_json' not in str(item)):
                            text = item.get('text', '')
                            if text and not text.startswith('{"'):
                                # Add space after tool completion if needed
                                if self.last_was_tool:
                                    print(" ", end="", flush=True)
                                    self.last_was_tool = False
                                print(text, end="", flush=True)
                                self.in_text_generation = True
                                
        elif event == "on_tool_start":
            if self.in_text_generation:
                print(f"\n🔧 {chunk.get('name', 'tool')}", end="", flush=True)
                self.in_text_generation = False
                
        elif event == "on_tool_end":
            print(" ✅", end="", flush=True)
            self.last_was_tool = True

Вызов агента

Наконец, вызовите агента и спросите его, кто внёс больше всего кода в ClickHouse:

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("anthropic:claude-sonnet-4-0", tools)
        
        handler = UltraCleanStreamHandler()        
        async for chunk in agent.astream_events(
            {"messages": [{"role": "user", "content": "Кто внёс наибольший вклад в код ClickHouse?"}]}, 
            version="v1"
        ):
            handler.handle_chunk(chunk)
            
        print("\n")

Вы должны увидеть ответ, аналогичный приведенному ниже:

Помогу вам определить, кто внёс наибольший вклад в код ClickHouse, изучив доступные базы данных и таблицы с данными о git-коммитах.
🔧 list_databases ✅ Вижу базу данных `git`, которая, вероятно, содержит информацию о git-коммитах. Изучим таблицы в этой базе данных:
🔧 list_tables ✅ Отлично! Таблица `clickhouse_commits` в базе данных git содержит данные о коммитах ClickHouse — всего 80 644 коммита. В этой таблице хранится информация о каждом коммите, включая автора, добавленные/удалённые строки, изменённые файлы и т. д. Выполним запрос к этой таблице, чтобы определить, кто внёс наибольший вклад в код по различным метрикам.
🔧 run_select_query ✅ Также посмотрим только на добавленные строки, чтобы определить, кто внёс больше всего нового кода:
🔧 run_select_query ✅ Согласно данным о git-коммитах ClickHouse, **Алексей Миловидов** внёс наибольший вклад в код ClickHouse по нескольким показателям:

## Ключевая статистика:

1. **Наибольшее количество изменённых строк**: Алексей Миловидов — **1 696 929 изменённых строк** (853 049 добавлено + 843 880 удалено)
2. **Наибольшее количество добавленных строк**: Алексей Миловидов — **853 049 добавленных строк**
3. **Наибольшее количество коммитов**: Алексей Миловидов — **15 375 коммитов**
4. **Наибольшее количество изменённых файлов**: Алексей Миловидов — **73 529 изменённых файлов**

## Топ-контрибьюторов по добавленным строкам:

1. **Алексей Миловидов**: 853 049 добавленных строк (15 375 коммитов)
2. **s-kat**: 541 609 добавленных строк (50 коммитов) 
3. **Николай Кочетов**: 219 020 добавленных строк (4 218 коммитов)
4. **alesapin**: 193 566 добавленных строк (4 783 коммита)
5. **Виталий Баранов**: 168 807 добавленных строк (1 152 коммита)

Алексей Миловидов, безусловно, является наиболее продуктивным контрибьютором ClickHouse, что вполне логично, поскольку он один из создателей и ведущих разработчиков проекта. Его вклад значительно превосходит вклад остальных как по общему объёму кода, так и по количеству коммитов: почти 16 000 коммитов и более 850 000 строк кода, добавленных в проект.