ClickHouse MCP server を使用して LangChain/LangGraph AIエージェントを構築する方法

このガイドでは、LangChain/LangGraph AIエージェントを構築し、 ClickHouse の SQL playground を ClickHouse の MCP server 経由で操作する方法を学びます。

Example notebookこのサンプルは、examples リポジトリに notebook として用意されています。

前提条件

システムに Python がインストールされている必要があります。
システムに pip がインストールされている必要があります。
Anthropic の API キー、または他の LLM provider の API キーが必要です。

以下の手順は、Python REPL でもスクリプトでも実行できます。

ライブラリのインストール

次のコマンドを実行して、必要なライブラリをインストールします。

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

認証情報を設定する

次に、Anthropic APIキーを指定する必要があります。

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

Response

Enter Anthropic API Key: ········

別のLLMプロバイダーを使用するAnthropic APIキーをお持ちでなく、別のLLMプロバイダーを使用したい場合は、 Langchain Providers docsで認証情報の設定手順を確認できます。

MCPサーバーを初期化する

次に、ClickHouse MCP serverがClickHouse SQL playgroundを参照するように設定します。

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", {})
            
            # 実際のテキストコンテンツのみを処理し、ツール呼び出しストリームはスキップする
            if hasattr(chunk_data, 'content'):
                content = chunk_data.content
                if isinstance(content, str) and not content.startswith('{"'):
                    # 必要に応じてツール完了後にスペースを追加する
                    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('{"'):
                                # 必要に応じてツール完了後にスペースを追加する
                                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": "Who's committed the most code to ClickHouse?"}]}, 
            version="v1"
        ):
            handler.handle_chunk(chunk)
            
        print("\n")

以下のような応答が表示されるはずです。

Response

利用可能なデータベースとテーブルを調査してgitコミットデータを特定し、ClickHouseに最も多くのコードをコミットした人物を見つけるお手伝いをします。
🔧 list_databases ✅ `git` データベースが存在することを確認しました。これにはgitコミット情報が含まれている可能性があります。そのデータベース内のテーブルを調べてみましょう：
🔧 list_tables ✅ 完璧です！gitデータベース内の `clickhouse_commits` テーブルに、80,644件のコミットを含むClickHouseのコミットデータが格納されていることを確認しました。このテーブルには、著者、追加/削除された行数、変更されたファイルなど、各コミットに関する情報が含まれています。さまざまなメトリクスに基づいて最も多くのコードをコミットした人物を特定するため、このテーブルにクエリを実行してみましょう。
🔧 run_select_query ✅ 追加された行数だけを見て、最も多くの新しいコードを貢献した人物も確認してみましょう：
🔧 run_select_query ✅ ClickHouseのgitコミットデータに基づくと、**Alexey Milovidov** が複数の指標においてClickHouseに最も多くのコードをコミットしています：

## 主要な統計：

1. **変更された総行数が最多**：Alexey Milovidov — **1,696,929行**（853,049行追加 + 843,880行削除）
2. **追加された行数が最多**：Alexey Milovidov — **853,049行追加**
3. **コミット数が最多**：Alexey Milovidov — **15,375コミット**
4. **変更されたファイル数が最多**：Alexey Milovidov — **73,529ファイル変更**

## 追加行数によるトップコントリビューター：

1. **Alexey Milovidov**：853,049行追加（15,375コミット）
2. **s-kat**：541,609行追加（50コミット）
3. **Nikolai Kochetov**：219,020行追加（4,218コミット）
4. **alesapin**：193,566行追加（4,783コミット）
5. **Vitaly Baranov**：168,807行追加（1,152コミット）

Alexey Milovidovは明らかにClickHouseへの最大のコントリビューターであり、プロジェクトの創設者の一人かつリード開発者であることを考えると納得できます。彼の貢献はコードの総量・コミット数ともに他を大きく引き離しており、約16,000件のコミットと85万行以上のコードがプロジェクトに追加されています。

​前提条件

​ライブラリのインストール

​認証情報を設定する

​MCPサーバーを初期化する

​ストリームハンドラーを設定する

​エージェントに問い合わせる

前提条件

ライブラリのインストール

認証情報を設定する

MCPサーバーを初期化する

ストリームハンドラーを設定する

エージェントに問い合わせる