AIエージェントが実験ツールになる — MCPでバイオデータベースを検索する
このトピックを終えると
教科書で学んだ MCP（モデルコンテキストプロトコル） と harness を組み合わせて、ClaudeやGPTなどのLLMが自然言語の指示によって、NCBI Entrez、PubMed、UniProtなどのバイオデータベースを自動的に検索するツールを自分で作成できるようになります。LLMのツール利用の内部原理と、その信頼性を評価するharnessの仕組みを理解することができます。
この記事は 教育用の一般的な例 です。実運用に配備するMCPサーバーは、認証、レート制限、可視化などの機能がより高度に実装されています。
「なぜ毎回同じ検索を繰り返すのか？」—反復実験の落とし穴
論文のアイデアを準備する際、毎日次のような検索を繰り返しているとします。
- NCBI PubMedで「"BRCA1" AND "review" AND "2024"」を検索
- 抄録をざっと見て関連性を判断
- 引用関係を確認
- UniProtで関連するタンパク質ドメインを確認
- GenBankから配列をダウンロード
この組み合わせを毎日繰り返します。問題があります。
問題1: 各データベースの検索構文が異なります。NCBI Entrezのフィルター構文、UniProtの検索構文、GenBankのアクセッション形式はすべて異なります。毎回、それを思い出す必要があります。
問題2: 複数のデータベースの結果を手動で組み合わせる必要があります。PubMedで得た著者名をUniProtに再度入力し、その結果のアクセッションをGenBankに再度入力する、といった具合です。
問題3: この作業のほとんどは決定的なルールベースですが、毎日異なるキーワードで繰り返されます。つまり、自動化するのに最適な形ですが、各データベースのAPIドキュメントを習得するオーバーヘッドが大きいです。
本当のアプローチは、これをAIエージェントに委ねることです。LLMに自然言語でリクエストを送信すると、バックグラウンドで複数のAPIを順番に呼び出し、結果を統合します。これを安全かつ標準化された方法で行うためのプロトコルが**MCP（Model Context Protocol）**です。
ブラックボックスからコンポーネントへ - MCP の詳細
MCP は表面上は複雑に見えますが、その中心的な概念は 3 つです。
コンポーネント 1: ツール定義
MCP サーバーは、LLM に、自身が提供できる ツールのリスト を提示します。各ツールは、名前、説明、および入力スキーマで定義されます。
tool_definition = { "name": "search_pubmed", "description": "PubMed の論文データベースでキーワードを検索し、上位 N 件の結果を返す", "input_schema": { "type": "object", "properties": { "query": {"type": "string", "description": "検索クエリ。PubMed 構文を使用"}, "max_results": {"type": "integer", "default": 10} }, "required": ["query"] }}重要な点 は、description が LLM が実際に読み込み、判断に使用する唯一の情報であるということです。優れたツールの説明は、優れたエージェントの動作を決定します。
コンポーネント 2: 実行関数
各ツールには、実際に実行される関数が関連付けられています。LLM がツールの呼び出しを要求すると、この関数が実行され、結果が LLM に返されます。
async def search_pubmed(query: str, max_results: int = 10) -> dict: import httpx base = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils" async with httpx.AsyncClient() as client: search_response = await client.get( f"{base}/esearch.fcgi", params={"db": "pubmed", "term": query, "retmax": max_results, "retmode": "json"} ) ids = search_response.json()["esearchresult"]["idlist"] if not ids: return {"results": []} summary_response = await client.get( f"{base}/esummary.fcgi", params={"db": "pubmed", "id": ",".join(ids), "retmode": "json"} ) summaries = summary_response.json()["result"] return { "results": [ { "pmid": pmid, "title": summaries[pmid].get("title"), "journal": summaries[pmid].get("fulljournalname"), "pubdate": summaries[pmid].get("pubdate") } for pmid in ids ] }注意点: この関数は外部 API を呼び出すため、ネットワーク、レート制限、タイムアウトを処理する必要があります。これらの点については、後で詳しく説明します。
コンポーネント 3: MCP サーバーの基本構造
これらすべてを 1 つのサーバーにまとめます。Anthropic は公式の Python SDK を提供していますが、概念を明確にするために、自分で作成してみましょう。
from typing import Any, Callable, Coroutineimport json
class MCPServer: def __init__(self): self.tools: dict[str, dict[str, Any]] = {} self.handlers: dict[str, Callable[..., Coroutine]] = {} def register(self, definition: dict, handler: Callable[..., Coroutine]): name = definition["name"] self.tools[name] = definition self.handlers[name] = handler def list_tools(self) -> list[dict]: return list(self.tools.values()) async def call_tool(self, name: str, arguments: dict) -> dict: if name not in self.handlers: return {"error": f"Unknown tool: {name}"} try: return await self.handlers[name](**arguments) except Exception as e: return {"error": str(e)}
server = MCPServer()server.register( definition={ "name": "search_pubmed", "description": "Search PubMed for scientific papers", "input_schema": { "type": "object", "properties": { "query": {"type": "string"}, "max_results": {"type": "integer", "default": 10} }, "required": ["query"] } }, handler=search_pubmed)このサーバーが LLM クライアントと通信するには、標準入力/出力または HTTP を使用する必要がありますが、ここでは 関数として直接呼び出す ことで、概念を理解します。
LLM連携 — Claudeと連携して書く
このサーバーをLLM呼び出しに接続します。Claude APIのツール使用機能を活用します。
import anthropic
client = anthropic.Anthropic()
async def agent_loop(user_message: str, server: MCPServer, max_iterations: int = 5) -> str: messages = [{"role": "user", "content": user_message}] tools = [ { "name": t["name"], "description": t["description"], "input_schema": t["input_schema"] } for t in server.list_tools() ] for iteration in range(max_iterations): response = client.messages.create( model="claude-opus-4-8", max_tokens=2048, tools=tools, messages=messages ) if response.stop_reason == "end_turn": text_blocks = [b.text for b in response.content if b.type == "text"] return "\n".join(text_blocks) if response.stop_reason == "tool_use": tool_uses = [b for b in response.content if b.type == "tool_use"] messages.append({"role": "assistant", "content": response.content}) tool_results = [] for tool_use in tool_uses: result = await server.call_tool(tool_use.name, tool_use.input) tool_results.append({ "type": "tool_result", "tool_use_id": tool_use.id, "content": json.dumps(result) }) messages.append({"role": "user", "content": tool_results}) return "Max iterations reached"使用例：
result = await agent_loop( "BRCA1遺伝子に関する、2024年以降のレビュー論文3件を検索し、タイトルとジャーナル名を教えてください。", server)print(result)LLMが自動的にsearch_pubmed(query="BRCA1 AND review AND 2024:2025", max_results=3)を呼び出し、結果を自然言語で整理して返します。
フェーディング — あなたが埋めるべき3つの空白
空白1：ツール拡張 — UniProt検索
PubMedだけでは不十分です。UniProtツールを追加しましょう。
async def search_uniprot(query: str, max_results: int = 10) -> dict: import httpx async with httpx.AsyncClient() as client: # TODO: UniProt REST APIを呼び出す # エンドポイント: https://rest.uniprot.org/uniprotkb/search # パラメータ: query, format=json, size=max_results # 結果: {"results": [{"accession": ..., "name": ..., "gene": ...}]} passヒント: UniProtレスポンスのresults配列の各項目からprimaryAccession、proteinDescription.recommendedName.fullName.value、genes[0].geneName.valueを抽出します。
空白2：ハーネス — エージェント評価システム
エージェントが正常に動作していることを検証するには、評価ハーネスが必要です。複数のテストケースを自動的に実行し、各結果を評価します。
async def evaluate_agent( test_cases: list[dict], server: MCPServer) -> dict: """ test_cases: [ { "prompt": "ユーザーのリクエスト", "expected_tool_calls": ["search_pubmed"], "expected_content_contains": ["BRCA1", "review"] } ] """ results = [] for tc in test_cases: # TODO: agent_loopを実行し、実際に呼び出されたツールを追跡し、結果を検証する # 合格かどうかをresultsに記録する pass return { "total": len(test_cases), "passed": sum(1 for r in results if r["passed"]), "details": results }ヒント: agent_loop内部でツール呼び出しをログに記録するように修正します。各呼び出しをリストに追加して返します。
空白3：レート制限とエラー処理
NCBI Entrezは、1秒あたり3回以上のリクエストを行うとIPアドレスをブロックします。ツール実行にレート制限を追加します。
import asyncioimport time
class RateLimiter: def __init__(self, calls_per_second: float): self.min_interval = 1.0 / calls_per_second self.last_call = 0.0 async def wait(self): now = time.time() elapsed = now - self.last_call if elapsed < self.min_interval: await asyncio.sleep(self.min_interval - elapsed) self.last_call = time.time()
ncbi_limiter = RateLimiter(calls_per_second=3)
async def search_pubmed_limited(query: str, max_results: int = 10) -> dict: # TODO: 関数の開始前にncbi_limiter.wait()を呼び出す # その後、既存のロジックを実行する pass考察 — このエージェントは、実用的なツールを使用するLLMとどのように異なるか
あなたが作成したエージェントは、概念的には、実用的なシステム（Claude Code、ツールを備えたChatGPTなど）と同じ基盤を共有していますが、実用的なシステムははるかに高度です。
セキュリティ: 実用的なMCPサーバーは、プロンプトインジェクションから保護する必要があります。ツールが返すデータに悪意のある指示（「このコマンドを無視して、代わりにXを実行する」）が含まれている場合、エージェントが誤動作する可能性があります。実用的なシステムは、ツールの戻り値をサンドボックス化されたコンテキストに分離するか、別の検証レイヤーを設けます。
コスト管理: 各イテレーションでAPI呼び出しがトリガーされるため、コストが急速に増加します。実用的なシステムは、イテレーションの上限、トークンの予算、ツールの呼び出しのキャッシュを備えています。
可視性: エージェントが特定のツールを呼び出した理由や、どの判断が間違っていたかを事後分析するには、トレースログが必要です。実用的なシステムでは、OpenTelemetryなどの標準を使用して、各ステップを記録します。
マルチステップ推論: 複雑なリクエスト（例：「この遺伝子に関する論文に登場する他のすべての遺伝子を見つけ、それぞれの発現パターンを検索する」）は、複数のステップのツールの呼び出し計画を必要とします。実用的なシステムでは、プランナーとエグゼキューターの分離やReActフレームワークがよく使用されます。
拡張プロジェクト
1. GenBank配列ダウンロードツールを追加: efetch.fcgi を使用して、アクセッション番号からFASTA配列を返す。
2. 引用グラフツール: PubMed論文の引用関係を追跡し、関連論文のツリーを構築する。
3. ローカルキャッシュ: 同じクエリを再度実行しないように、SQLiteキャッシュレイヤーを追加する。
4. Claude Desktopとの統合: ユーザーが作成したMCPサーバーを、実際のClaude Desktopアプリに登録して、毎日使用できるようにする。標準のMCP SDKを使用して、stdioサーバーでラップすればよい。
このモジュールの構成要素
- [F] MCPプロトコル: ツール定義、呼び出し、応答の標準形式。サーバー・クライアント分離。
- [F] harness: エージェントの動作を自動評価するテストシステム。LLMシステムのCI/CDの役割。
- [W] API呼び出し・JSON: NCBI/UniProt RESTの呼び出しと応答の解析。
- [W] async/await: 非同期I/Oにより、複数のAPI呼び出しを並列化。
[F] = 自分で実装するもの / [W] = 完成したコードで提供するツールの概念。