本記事の目標

本シリーズの第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でのデバッグ方法について解説します。