LoginSignup
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@BNR-Longin株式会社BinaryTech | 日本品質 × ベトナム開発、最高のバランスを。

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

Posted at

はじめに

第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にアクセス。
    • アカウントを作成し、APIキーを取得（無料枠では制限あり、有料プラン推奨）。
    • APIキーを環境変数BRAVE_API_KEYに保存。
  • ライブラリのインストール： 
    pip install requests

2. Qdrantの設定

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

3. 環境変数の設定

以下の環境変数を.envファイルまたはシステムに設定：

export BRAVE_API_KEY=your_brave_api_key
export QDRANT_URL=http://localhost:6333
export QDRANT_COLLECTION=ai_news

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

pip install python-dotenv

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

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

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. サーバー起動：

    python realtime_data_server.py

    コンソールに「リアルタイムデータMCPサーバーを起動中: http://localhost:8090」と表示。

  2. ウェブ検索のテスト：
    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))

    期待されるレスポンス：

    {
  "jsonrpc": "2.0",
  "result": {
    "status": "success",
    "results": [
      {
        "title": "AIが医療を変革",
        "url": "https://example.com/ai-health",
        "description": "AIが診断精度を向上..."
      }
    ]
  },
  "id": 1
}

  3. Qdrantデータのテスト：

    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）に以下を追加：

    {
  "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コストを削減。

簡略化したロジック例：

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の未来に興味がある方は、ぜひ最終章をお楽しみに！

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

1
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?