MCPサーバー構築 FastMCP 並列検索 Grok API Gemini Search Brave Search

🤔 なぜ検索エンジン1つでは足りないの？

AIエージェントを作っていると、ウェブ検索が必須になる瞬間が来ます。最新ニュースを調べたり、トレンドを調査したり、技術ドキュメントを確認するとき — 検索なしではハルシネーション（幻覚）が増えるばかりですよね。でも、検索エンジンを1つだけ使うとバイアスが生まれます。GoogleはSEOに最適化された結果が多いし、Braveは独立インデックスで違うソースを見せてくれるし、GrokはX（Twitter）のリアルタイムデータまで取得できるんです。

そこで、3つの検索ソースを同時に呼び出し、結果をクロス検証して総合レポートを作るMCPサーバーを自分で作りました。名前はterry-research。MCPサーバー構築に興味があるなら、今日は設計からasyncioバグの解決、実践テストまで正直にシェアしますね。

🏗️ 全体アーキテクチャ — オーケストレーターパターン

terry-researchの核心はオーケストレーター（Orchestrator）パターンです。1人の指揮者が複数の検索エージェントに同時に仕事を振り分けて、結果を集めて統合する構造です。

全体で1,066行。小さなコードベースですが、エージェント分離と並列処理のおかげで拡張しやすい構造です。

🔧 4つのMCPツール

terry-researchは4つのツールを提供します。research()がメインで、残りの3つは個別ソースを直接呼び出すときに使います。

@mcp.tool()
def research(
    query: str,
    topic_type: str = "general",
    depth: str = "standard",
    sources: list[str] | None = None,
    include_images: bool = False,
    language: str = "ko",
    max_results_per_source: int = 10,
) -> str:
    """Multi-source research
    with parallel dispatch."""

🎯 トピックルーティング — テーマに合うソースを自動選択

research()を呼び出すときtopic_typeを指定すると、config.yamlのルーティングテーブルに従って最適なソース組み合わせが自動的に選択されます。

# config.yaml
topic_routing:
  general:
    - grok
    - gemini
  fashion_trend:
    - grok
    - gemini
  sns_trend:
    - grok          # X（Twitter）データ
  tech:
    - gemini
    - brave         # 独立インデックス
  news:
    - grok
    - gemini

例えば、SNSトレンドはGrokだけを使います。Xのリアルタイム投稿を直接検索できるのはGrokだけだからです。技術テーマにはGemini + Braveの組み合わせがいいですね — Googleの膨大なインデックスとBraveの独立した視点を一緒に見られます。

depthパラメータで検索の深さも調整できます。

deepモードでは、Grokがgrok-4-1-fast-reasoningに、Geminiがgemini-3.1-pro-previewにアップグレードされ、検索ターン数も3から10に増えます。

⚡ 並列ディスパッチ — ThreadPoolExecutor

一番悩んだ部分です。3つのAPIを同時に呼び出す必要があるので、非同期処理は必須でした。最初は当然asyncioを使いました。

# 最初の試み（失敗！）
async def _dispatch(self, sources):
    tasks = [
        self._search(s) for s in sources
    ]
    return await asyncio.gather(*tasks)

解決策はThreadPoolExecutorでした。スレッドベースの並列処理はイベントループと衝突しません。

def _dispatch_parallel(
    self, query, sources, deep, lang
) -> list[SourceResponse]:
    with ThreadPoolExecutor(
        max_workers=len(sources)
    ) as pool:
        futures = {
            pool.submit(
                self._search_single,
                src, query, deep, lang
            ): src
            for src in sources
        }
        results = []
        for fut in as_completed(
            futures, timeout=self.timeout
        ):
            try:
                results.append(fut.result())
            except Exception as e:
                src = futures[fut]
                results.append(
                    SourceResponse(
                        source=src,
                        error=str(e)
                    )
                )
        return results

これで各検索エージェントが別々のスレッドで独立して実行されます。1つのソースが遅くても他には影響せず、タイムアウト（デフォルト30秒）を過ぎたらまだ終わっていないソースはスキップされます。

🔀 結果マージ — クロス検証の力

3つのソースから結果が返ってくると、Mergerが重複を除去してクロス検証します。

def merge(self, responses):
    # 1. URL正規化（www、末尾スラッシュ除去）
    # 2. 引用の重複除去
    # 3. multi_source_urls追跡
    #    — 2つ以上のソースに出現したURL
    # 4. per_source個別回答を保持

ポイントはmulti_source_urlsです。同じURLがGrokとGeminiの両方に出てきたら、その情報の信頼度は高いということです。レポートに「クロス検証済み」の表示がつくので、AIクライアントがより信頼できます。

🧩 エージェントごとの特徴

各エージェントは、そのAPIの強みを最大限活用するように設計しました。

Grokの最大の強みはXの投稿まで検索できること。SNSトレンド調査でリアルタイムの反応を見られます。Geminiはgrounding_metadataから検索根拠をURI/タイトルのペアで返してくれるので引用が正確です。BraveはGoogleと重ならない独立インデックスで多様性を確保できます。

📊 実践テスト結果

「2026年韓国ファッショントレンド」をtopic_type: fashion_trendで検索した結果です。

33秒で24件の引用を含む10,724文字のレポートが出来上がります。各ソースを順番に呼び出していたら60秒以上かかっていたでしょうが、並列ディスパッチのおかげでほぼ半分に短縮されました。

🛠️ 使ってみよう — セットアップガイド

自分でMCPサーバー構築に挑戦したいなら、terry-researchの構造が良い出発点です。GitHubで公開予定で、Claude CodeやCursorに登録すればすぐに使えます。

# 1. クローン（公開後）
git clone https://github.com/goandon/terry-research.git
cd terry-research

# 2. 依存関係インストール
pip install -r requirements.txt

# 3. APIキー設定（.env）
XAI_API_KEY=your-xai-key
GEMINI_API_KEY=your-gemini-key
BRAVE_API_KEY=your-brave-key

# 4. MCPサーバー起動
python server.py

APIキーがないソースは自動的にスキップされます。Braveのキー1つだけでもbrave_search()ツールは正常に動作します。

{
  "terry-research": {
    "command": "python",
    "args": ["server.py"],
    "cwd": "/path/to/terry-research"
  }
}

💡 作りながら学んだこと

• FastMCP + asyncioの組み合わせは罠です — FastMCP内部でイベントループがすでに動いているので、並列処理にはThreadPoolExecutorが安全です
• エラー分離は必須です — 外部APIを3つも呼ぶと、1つは必ず問題が起きます。ソースごとのエラーハンドリングがないと全体が崩壊します
• クロス検証こそ真の価値 — 複数のソースから同じURLが出てくると信頼度が上がります。単に結果を結合するのとは次元が違います
• トピックルーティングでコスト削減 — 毎回全ソースを呼ぶとコストと時間が無駄です。テーマに合ったソースだけを選ぶのが実用的です
• 1,066行で十分です — 過度に抽象化せず、エージェント1つにつきファイル1つを維持すれば、読みやすくメンテもしやすいです

📚 参考資料

• terry-research GitHub — 全ソースコードとセットアップガイド（公開予定）
• Model Context Protocol 公式サイト — MCPスペックとSDKドキュメント
• FastMCP GitHub — Python MCPサーバーフレームワーク
• xAI APIドキュメント — Grok Responses API + 検索ツール
• Gemini APIドキュメント — Google Searchグラウンディング
• Brave Search API — 独立インデックス検索API
• BookStack Wikiセルフホスティングガイド — 同じインフラシリーズ（関連記事）
• Synology NAS SSL設定ガイド — MCPサーバーホスティングに活用可能（関連記事）

✅ まとめ

terry-research MCPの核心は3つです — トピックルーティングで最適なソース選択、ThreadPoolExecutorで並列ディスパッチ、URLクロス検証で信頼性確保。このパターンは検索だけでなく、複数のAPIを同時に呼び出す必要があるすべてのMCPサーバーに応用できます。

自分でMCPサーバー構築に挑戦してみると、AIエージェントの可能性が大きく広がるのを実感できます。検索を1つ追加するだけでもエージェントがずっと賢くなりますよ。皆さんもぜひ、自分だけのMCPサーバーに挑戦してみてください！