MCPとエージェントAI：自律型AIを構築する | 第4章：リアルタイムデータの活用

## はじめに

第3章では、**Model Context Protocol（MCP）**を使ってNotionとSlackを統合し、プロジェクト管理を自動化するエージェントAIを構築しました。Slackのメッセージからタスクを抽出し、Notionに追加する機能により、チームのワークフローを効率化できました。今回は、エージェントAIをさらに進化させ、**リアルタイムデータ**を取り入れる方法を学びます。具体的には、**Brave Search**と**Qdrant**をMCPサーバーに統合し、最新情報やベクトルデータに基づく回答を提供するエージェントを構築します。

この第4章では、Brave Searchでウェブを検索し、Qdrantでベクトルデータを管理するMCPサーバーを作成します。たとえば、「最新のAIニュースを教えて」と依頼すると、エージェントがBrave Searchから情報を取得し、Qdrantに保存して要約を返します。コード例とステップごとのガイドで、リアルタイムデータの可能性を体感しましょう。さあ、始めましょう！

## 準備：Brave SearchとQdrantのセットアップ

Brave SearchとQdrantをMCPサーバーに統合するには、両方のサービスを設定する必要があります。以下の手順で準備します：

### 1. Brave Search APIの設定
- **APIキーの取得**：
  - [Brave Search API](https://api.search.brave.com/)にアクセス。
  - アカウントを作成し、APIキーを取得（無料枠では制限あり、有料プラン推奨）。
  - APIキーを環境変数`BRAVE_API_KEY`に保存。
- **ライブラリのインストール**：
  ```bash
  pip install requests
  ```

### 2. Qdrantの設定
- **Qdrantの起動**：
  - ローカルでQdrantをDockerで起動：
    ```bash
    docker run -p 6333:6333 qdrant/qdrant
    ```
  - または、[Qdrant Cloud](https://cloud.qdrant.io/)でホストされたインスタンスを使用。
- **コレクションの作成**：
  - Qdrantクライアントを使って、検索結果を保存するコレクションを作成（後述のコードで自動化）。
- **ライブラリのインストール**：
  ```bash
  pip install qdrant-client
  ```

### 3. 環境変数の設定
以下の環境変数を`.env`ファイルまたはシステムに設定：
```bash
export BRAVE_API_KEY=your_brave_api_key
export QDRANT_URL=http://localhost:6333
export QDRANT_COLLECTION=ai_news
```

ライブラリ`python-dotenv`で環境変数を読み込む：
```bash
pip install python-dotenv
```

## MCPサーバーの構築：Brave SearchとQdrantの統合

以下のMCPサーバーは、Brave Searchでウェブを検索し、結果をQdrantに保存する機能を提供します。エージェントAIは、保存されたデータを参照して、ユーザーの質問に最新情報を基にした回答を生成します。

```python
from mcp import MCPServer
from qdrant_client import QdrantClient
from qdrant_client.http.models import Distance, VectorParams, PointStruct
import requests
import os
from dotenv import load_dotenv
from datetime import datetime

class RealTimeDataServer(MCPServer):
    def __init__(self, host, port, brave_api_key, qdrant_url, qdrant_collection):
        super().__init__(host, port)
        self.brave_api_key = brave_api_key
        self.qdrant_client = QdrantClient(qdrant_url)
        self.collection_name = qdrant_collection
        self._init_qdrant_collection()
        self.register_resource("search_web", self.search_web)
        self.register_resource("get_stored_results", self.get_stored_results)

    def _init_qdrant_collection(self):
        try:
            collections = self.qdrant_client.get_collections()
            if self.collection_name not in [c.name for c in collections.collections]:
                self.qdrant_client.create_collection(
                    collection_name=self.collection_name,
                    vectors_config=VectorParams(size=768, distance=Distance.COSINE)
                )
        except Exception as e:
            print(f"コレクション作成エラー: {str(e)}")

    def search_web(self, params):
        query = params.get("query", "")
        try:
            url = "https://api.search.brave.com/res/v1/web/search"
            headers = {"X-Subscription-Token": self.brave_api_key}
            params = {"q": query, "count": 5}
            response = requests.get(url, headers=headers, params=params)
            response.raise_for_status()
            results = response.json().get("web", {}).get("results", [])
            formatted_results = [
                {"title": r["title"], "url": r["url"], "description": r["description"]}
                for r in results
            ]
            # Qdrantに保存
            for i, result in enumerate(formatted_results):
                self.qdrant_client.upsert(
                    collection_name=self.collection_name,
                    points=[
                        PointStruct(
                            id=f"{query}_{i}",
                            vector=[0.1] * 768,  # ダミーベクトル（実際はBERT等で生成）
                            payload={
                                "title": result["title"],
                                "url": result["url"],
                                "description": result["description"],
                                "timestamp": datetime.utcnow().isoformat()
                            }
                        )
                    ]
                )
            return {"status": "success", "results": formatted_results}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def get_stored_results(self, params):
        query = params.get("query", "")
        try:
            results = self.qdrant_client.scroll(
                collection_name=self.collection_name,
                scroll_filter={"must": [{"key": "title", "match": {"text": query}}]},
                limit=5
            )
            formatted_results = [point.payload for point in results[0]]
            return {"status": "success", "results": formatted_results}
        except Exception as e:
            return {"status": "error", "message": str(e)}

if __name__ == "__main__":
    load_dotenv()
    server = RealTimeDataServer(
        host="localhost",
        port=8090,
        brave_api_key=os.getenv("BRAVE_API_KEY"),
        qdrant_url=os.getenv("QDRANT_URL"),
        qdrant_collection=os.getenv("QDRANT_COLLECTION")
    )
    print("リアルタイムデータMCPサーバーを起動中: http://localhost:8090")
    server.start()
```

### コードの説明
- **_init_qdrant_collection**：Qdrantにコレクションを作成（ベクトルサイズ768は仮定、実際はBERTモデル等で調整）。
- **search_web**：Brave Searchでクエリを検索し、結果をフォーマット。Qdrantに保存。
- **get_stored_results**：Qdrantから保存済みの検索結果を取得。
- **register_resource**：リソースを登録し、AIがアクセス可能に。

### 前提条件
- Brave Search APIキーが有効。
- Qdrantがローカルまたはクラウドで稼働中。
- ベクトル生成は簡略化（実際にはHugging Faceの`sentence-transformers`等を使用）。

## サーバーのテスト

サーバーが正しく動作するか確認します：

1. **サーバー起動**：
   ```bash
   python realtime_data_server.py
   ```
   コンソールに「リアルタイムデータMCPサーバーを起動中: http://localhost:8090」と表示。

2. **ウェブ検索のテスト**：
   Pythonでリクエストを送信：
   ```python
   import requests
   import json

   url = "http://localhost:8090"
   payload = {
       "jsonrpc": "2.0",
       "method": "search_web",
       "params": {"query": "最新のAIニュース"},
       "id": 1
   }
   response = requests.post(url, json=payload)
   print(json.dumps(response.json(), indent=2, ensure_ascii=False))
   ```
   期待されるレスポンス：
   ```json
   {
     "jsonrpc": "2.0",
     "result": {
       "status": "success",
       "results": [
         {
           "title": "AIが医療を変革",
           "url": "https://example.com/ai-health",
           "description": "AIが診断精度を向上..."
         }
       ]
     },
     "id": 1
   }
   ```

3. **Qdrantデータのテスト**：
   ```python
   payload = {
       "jsonrpc": "2.0",
       "method": "get_stored_results",
       "params": {"query": "AI"},
       "id": 2
   }
   response = requests.post(url, json=payload)
   print(json.dumps(response.json(), indent=2, ensure_ascii=False))
   ```

## Claude Desktopとの統合

サーバーをClaude Desktopに接続します：

1. **設定ファイルの編集**：
   Claude Desktopの設定ファイル（例：`claude_desktop_config.json`）に以下を追加：
   ```json
   {
     "mcp_servers": [
       {
         "name": "RealTimeDataServer",
         "url": "http://localhost:8090",
         "auth": "none"
       }
     ]
   }
   ```

2. **Claudeでテスト**：
   Claude Desktopを起動し、プロンプトを入力：
   ```
   最新のAIニュースを教えてください。
   ```
   レスポンス例：
   ```
   Brave Searchから取得した最新のAIニュース：
   - タイトル：AIが医療を変革
     URL：https://example.com/ai-health
     概要：AIが診断精度を向上...
   ```

   保存データのプロンプト：
   ```
   保存済みのAI関連ニュースを教えて。
   ```
   レスポンス例：
   ```
   Qdrantに保存されたニュース：
   - タイトル：AIが医療を変革
     URL：https://example.com/ai-health
     取得日：2025-04-15
   ```

## エージェントロジックの拡張

エージェントAIがリアルタイムデータを効果的に活用するには、以下のようなロジックを追加します：
- **自動更新**：定期的にBrave Searchを呼び出し、Qdrantを最新情報で更新。
- **要約生成**：検索結果の概要をClaudeに要約させるプロンプトを追加。
- **キャッシュ戦略**：頻繁なクエリに対してQdrantを優先し、APIコストを削減。

簡略化したロジック例：
```python
def process_query(self, query):
    stored = self.get_stored_results({"query": query})["results"]
    if stored:
        return stored
    return self.search_web({"query": query})["results"]
```

## 実装のコツと注意点

- **API制限**：Brave Searchの無料枠はリクエスト数に制限あり。有料プランを検討。
- **ベクトル生成**：本番では、`sentence-transformers`を使って検索結果の埋め込みを生成。
- **エラーハンドリング**：API障害時に代替データをQdrantから取得。
- **セキュリティ**：APIキーを環境変数で管理し、公開しない。
- **スケーラビリティ**：Qdrantのコレクションサイズが大きくなったら、パーティショニングを検討。

## 試してみよう：挑戦課題
以下の機能を追加して、エージェントを強化してみてください：
- Brave Searchのニュース専用エンドポイント（`/news/search`）を使用。
- Qdrantに保存する際に、検索結果の関連度スコアを追加。
- ユーザーが「要約して」と依頼したら、Claudeで結果を要約。

## まとめと次のステップ

この第4章では、Brave SearchとQdrantをMCPで統合し、リアルタイムデータに基づくエージェントAIを構築しました。ウェブ検索とベクトルデータベースを組み合わせることで、最新情報を提供する強力なエージェントが実現しました。

次の第5章では、エージェントAIの**セキュリティと倫理**に焦点を当て、MCPの安全な運用方法やコミュニティへの貢献について学びます。エージェントAIの未来に興味がある方は、ぜひ最終章をお楽しみに！

---

役に立ったと思ったら、「いいね」や「ストック」をしていただけると嬉しいです！次の章でまたお会いしましょう！