本記事の目標
本シリーズの第3回では、MCP（Model Context Protocol）を使って外部ツールと連携する方法を解説します。校正エージェントでは、技術的な正確性を検証するためにAWS Knowledge MCPやTavily検索を使用しています。
MCPはAnthropicが提唱するオープンプロトコルで、LLMと外部サービスを標準的な方法で接続できます。
なぜMCPを使うのか
技術記事の校正では、以下のような検証が必要です。
- APIの仕様が最新の情報と一致しているか
- コマンドのオプションが正確か
- 技術用語の説明が正しいか
これらを検証するには、外部の知識ベースやWeb検索が必要です。MCPを使うことで、以下のメリットがあります。
- 標準化されたインターフェース：異なるツールでも同じ方法で呼び出せる
- 非同期実行：I/O待ちの間に他の処理を進められる
- 動的なツール発見：利用可能なツールを実行時に取得できる
langchain-mcp-adaptersの使用
LangChainエコシステムでは、langchain-mcp-adaptersパッケージを使ってMCPサーバーに接続できます。
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_core.tools import BaseTool
AWS_KNOWLEDGE_MCP_URL = "https://knowledge-mcp.global.api.aws"
async def load_technical_search_tools(
enable_aws_mcp: bool = True,
enable_tavily: bool = True,
) -> list[BaseTool]:
tools: list[BaseTool] = []
if enable_aws_mcp:
try:
client = MultiServerMCPClient(
{
"aws_knowledge": {
"transport": "streamable_http",
"url": AWS_KNOWLEDGE_MCP_URL,
}
}
)
mcp_tools = await client.get_tools()
tools.extend(mcp_tools)
logger.info(f"Loaded {len(mcp_tools)} tools from AWS Knowledge MCP")
except Exception as e:
logger.warning(f"Failed to load AWS Knowledge MCP tools: {e}")
if enable_tavily:
# Tavilyツールの追加
...
return tools
MultiServerMCPClientを使うことで、複数のMCPサーバーに同時に接続できます。await client.get_tools()で利用可能なツールを取得し、LangChainのBaseTool形式で返されます。
ツールを使うサブグラフの構造
MCPツールを使うチェッカーは、以下のような複雑なグラフ構造になります。
START
↓
extract_claims（技術的主張の抽出）
↓
plan_search（検索計画）
↓ (条件分岐)
├─ tools（ツール実行）
│ ↓
│ └─→ plan_search（継続判定）
│
└─→ verify_and_generate（検証・Issue生成）
↓
END
この構造により、必要に応じて複数回の検索を実行できます。
TechnicalAccuracyCheckerの実装
class TechnicalAccuracyChecker:
MAX_SEARCH_ITERATIONS = 10
def __init__(
self,
llm: ChatGoogleGenerativeAI,
tools: list[BaseTool],
) -> None:
self.llm = llm
self.tools = tools
self.llm_with_tools = llm.bind_tools(tools) if tools else llm
def create_graph(self) -> CompiledStateGraph:
builder = StateGraph(TechnicalAccuracyState)
builder.add_node("extract_claims", self._extract_claims_node)
builder.add_node("plan_search", self._plan_search_node)
if self.tools:
builder.add_node("tools", ToolNode(self.tools))
builder.add_node("verify_and_generate", self._verify_and_generate_node)
builder.add_edge(START, "extract_claims")
builder.add_edge("extract_claims", "plan_search")
if self.tools:
builder.add_conditional_edges(
"plan_search",
self._route_after_plan,
["tools", "verify_and_generate"],
)
builder.add_conditional_edges(
"tools",
self._should_continue_search,
["plan_search", "verify_and_generate"],
)
else:
builder.add_edge("plan_search", "verify_and_generate")
builder.add_edge("verify_and_generate", END)
return builder.compile()
条件分岐の実装
add_conditional_edgesを使って、動的にエッジの遷移先を決定します。
def _route_after_plan(
self, state: TechnicalAccuracyState
) -> Literal["tools", "verify_and_generate"]:
"""plan_searchノードの後の遷移先を決定"""
messages = state.get("messages", [])
if not messages:
return "verify_and_generate"
last_message = messages[-1]
# ツール呼び出しがあればtoolsノードへ
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
# なければverify_and_generateへ
return "verify_and_generate"
def _should_continue_search(
self, state: TechnicalAccuracyState
) -> Literal["plan_search", "verify_and_generate"]:
"""検索を継続するか判定"""
search_count = state.get("search_count", 0)
# 最大回数に達したら終了
if search_count >= self.MAX_SEARCH_ITERATIONS:
return "verify_and_generate"
# まだ続けられるならplan_searchへ
return "plan_search"
ToolNodeによるツール実行
LangGraphのToolNodeは、LLMが出力したツール呼び出しを実際に実行します。
from langgraph.prebuilt import ToolNode
if self.tools:
builder.add_node("tools", ToolNode(self.tools))
ToolNodeは以下の処理を自動で行います。
- メッセージからツール呼び出しを抽出
- 対応するツールを実行
- 結果を
ToolMessageとして返す
検索コンテキストの管理
複数回の検索結果を蓄積し、最終的な検証に活用します。
def _extract_search_context_from_tool_messages(
self, messages: list[BaseMessage]
) -> str:
"""ToolMessageから検索結果を抽出"""
context_parts = []
for msg in messages:
if isinstance(msg, ToolMessage):
try:
content = msg.content
if isinstance(content, str):
data = json.loads(content)
if isinstance(data, dict):
query = data.get("query", "Unknown query")
results = data.get("results", [])
context = f"### 検索クエリ: {query}\n"
for i, r in enumerate(results[:3], 1):
title = r.get("title", "No title")
snippet = r.get("content", "")
url = r.get("url", "")
context += f"{i}. **{title}**\n {snippet}\n URL: {url}\n\n"
context_parts.append(context)
except Exception as e:
logger.warning(f"Failed to extract context: {e}")
return "\n---\n".join(context_parts)
検索済みクエリの追跡
同じクエリを重複して検索しないよう、検索済みクエリを追跡します。
class TechnicalAccuracyState(TypedDict):
# ...
searched_queries: NotRequired[Annotated[list[str], add_values]]
async def _plan_search_node(self, state: TechnicalAccuracyState) -> dict[str, Any]:
searched_queries = state.get("searched_queries", [])
# プロンプトに検索済みクエリを含める
if searched_queries:
plan_prompt += "\n\n## 検索済みクエリ（これらは再検索しないでください）\n"
for q in searched_queries:
plan_prompt += f"- {q}\n"
response = await self.llm_with_tools.ainvoke(messages)
# 新しいクエリを抽出
new_queries = self._extract_queries_from_response(response)
return {
"messages": [response],
"search_count": search_count + 1,
"searched_queries": new_queries,
}
ファクトリ関数による初期化
MCPツールの読み込みは非同期処理のため、ファクトリ関数を使ってチェッカーを作成します。
async def create_technical_accuracy_checker(
llm: ChatGoogleGenerativeAI,
enable_aws_mcp: bool = True,
enable_tavily: bool = True,
) -> TechnicalAccuracyChecker:
tools = await load_technical_search_tools(
enable_aws_mcp=enable_aws_mcp,
enable_tavily=enable_tavily,
)
return TechnicalAccuracyChecker(llm, tools)
メイングラフでは以下のように使用します。
async def _build_graph(self) -> CompiledStateGraph:
# ...
technical_accuracy_checker = await create_technical_accuracy_checker(
self.llm,
enable_aws_mcp=self.config.enable_aws_mcp,
enable_tavily=self.config.enable_tavily,
)
builder.add_node(
"check_technical_accuracy",
technical_accuracy_checker.create_graph(),
)
# ...
エラーハンドリング
MCPサーバーへの接続に失敗した場合でも、エージェント全体が停止しないよう設計しています。
if enable_aws_mcp:
try:
client = MultiServerMCPClient(...)
mcp_tools = await client.get_tools()
tools.extend(mcp_tools)
except Exception as e:
# 接続失敗時はログを出して続行
logger.warning(f"Failed to load AWS Knowledge MCP tools: {e}")
ツールが一つも読み込めなかった場合は、検索なしで検証を行います。
まとめ
本記事では、MCPを使った外部ツール連携について解説しました。
ポイントは以下の通りです。
-
langchain-mcp-adaptersでMCPサーバーに接続 -
ToolNodeでツール呼び出しを自動実行 - 条件分岐で反復検索を実現
- 検索コンテキストを蓄積して最終検証に活用
- エラーハンドリングでMCP障害に対応
次回は、LangGraph Studioでのデバッグ方法について解説します。