MCP・Durable Execution・A2AでAIエージェント基盤を設計する実践ガイド

2026年、AIエージェントは「実験的なデモ」から「本番システムの構成要素」へと進化しました。Gartnerの予測によると、2026年末までに企業アプリケーションの40%がタスク固有のAIエージェントを組み込むとされています。しかし、エージェントを本番運用するためのインフラは、従来のWebアプリケーションとは根本的に異なる設計が求められます。

この記事では、2026年時点で確立されつつあるAIエージェント基盤の3つの柱 ── MCP（Model Context Protocol）、Durable Execution、A2A（Agent-to-Agent Protocol）── を中心に、実際の設計パターンと実装例を解説します。

この記事でわかること

• AIエージェント基盤を構成する3層プロトコルスタック（MCP・A2A・WebMCP）の役割と使い分け
• Durable Executionプラットフォーム（Temporal・Restate）を使ったエージェントの耐障害性設計
• Python / TypeScriptでMCPサーバーとエージェントオーケストレーションを実装する方法
• LangSmith・Langfuseによるエージェントオブザーバビリティの構築手順
• 各プラットフォーム（OpenAI・Anthropic・Google）のエージェントSDKの比較と選定基準

対象読者

• 想定読者: AIエージェントを本番環境にデプロイしたいMLエンジニア・バックエンドエンジニア
• 必要な前提知識:
  • Python 3.11+ または TypeScript 5.x の基本文法
  • LLM API（OpenAI API / Anthropic API）の基本的な使い方
  • Docker / コンテナの基礎知識
  • REST APIの設計・実装経験

結論・成果

本記事で解説するアーキテクチャパターンを採用することで、以下の効果が報告されています。

• 耐障害性: Temporal導入企業では、エージェントの中間障害からの自動回復により、手動リトライコードの実装が不要になったと報告されている（Temporal公式ブログ）
• 統合コスト: MCPの標準化により、ツール連携ごとのカスタムインテグレーション実装が不要になり、開発工数の削減が見込まれる
• オブザーバビリティ: LangSmithはトレーシングオーバーヘッドがほぼゼロで本番環境に導入可能（AIMultiple 2026調査）
• スケーラビリティ: Temporal Cloudは秒間150,000アクション以上のスパイクを事前通知なしで処理可能

AIエージェント基盤の3層プロトコルスタックを理解する

2026年2月時点で、AIエージェントのプロトコルアーキテクチャは3つの補完的なレイヤーに集約されつつあります（The 2026 AI Agent Protocol Wars Explained）。それぞれが異なる通信ニーズに対応しています。

MCP（Model Context Protocol）: エージェントとツールの接続

MCP は、Anthropicが2024年11月に発表したオープンプロトコルで、AIアプリケーションが外部のデータソース・ツール・サービスに接続するための標準インターフェースを提供します。Pythonでいえば __init__ のように、エージェントが起動時に利用可能なツールを自動検出する仕組みと考えてください。

MCPがなかった時代は、LLMと外部ツールを接続するたびにカスタムの関数定義・パース処理・エラーハンドリングを書く必要がありました。MCPはこれを標準化し、1つのプロトコルで任意のツールに接続できるようにします。

# mcp_server.py - FastMCPでシンプルなMCPサーバーを実装する例
# Python 3.11+, pip install mcp[server]
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("demo-tools")

@mcp.tool()
def search_database(query: str, limit: int = 10) -> list[dict]:
    """データベースからクエリに一致するレコードを検索する"""
    # 実際のDB接続処理（ここではサンプル）
    import sqlite3
    conn = sqlite3.connect("app.db")
    cursor = conn.execute(
        "SELECT * FROM documents WHERE content LIKE ? LIMIT ?",
        (f"%{query}%", limit)
    )
    results = [dict(zip([d[0] for d in cursor.description], row)) for row in cursor.fetchall()]
    conn.close()
    return results

@mcp.tool()
def create_ticket(title: str, description: str, priority: str = "medium") -> dict:
    """サポートチケットを作成する"""
    # チケット作成ロジック
    ticket_id = f"TICKET-{hash(title) % 10000:04d}"
    return {"id": ticket_id, "title": title, "status": "open", "priority": priority}

if __name__ == "__main__":
    mcp.run(transport="stdio")  # stdio経由でエージェントと通信

なぜMCPを選ぶのか:

• ベンダー非依存: OpenAI、Anthropic、Google、Microsoftの4大AIベンダーすべてがMCPを採用（日経クロステック）
• ツール再利用: 1つのMCPサーバーを複数のエージェントフレームワークから利用可能
• 自動検出: エージェントが接続時にツールのスキーマを自動取得し、関数呼び出しに変換

注意: MCPの認証機構はオプション仕様のため、本番環境ではOAuth 2.0やAPIキー管理を別途実装する必要があります。Anthropic公式のMCPコネクタはAPI側で接続管理を自動化しますが、セルフホストの場合はセキュリティ設計が必須です。

A2A（Agent-to-Agent Protocol）: エージェント間の協調

MCPが「エージェントとツール」の接続を担うのに対し、A2A は「エージェントとエージェント」の通信を標準化します。Googleが2025年4月に発表したオープンプロトコルで、異なるフレームワーク・ベンダーで構築されたエージェント同士が共通言語で協調できるようにします。

MCPとA2Aの使い分け:

観点	MCP	A2A
接続対象	エージェントとツール/データソース	エージェントとエージェント
通信パターン	リクエスト/レスポンス	非同期タスク委譲
状態管理	ステートレス（基本）	タスクライフサイクル管理
主なユースケース	DB検索、API呼び出し、ファイル操作	マルチエージェント協調、タスク分散
発表元	Anthropic（2024年11月）	Google（2025年4月）

よくある間違い: 「MCPだけですべてのエージェント間通信をカバーできる」と考えがちですが、MCPはツール接続に最適化されているため、エージェント同士のタスク委譲・進捗追跡・エラー伝播にはA2Aが適しています。単一エージェントの構成であればMCPだけで十分ですが、マルチエージェントシステムではA2Aの導入を検討してください。

WebMCP: Webコンテンツへの構造化アクセス

3層目の WebMCP は、Webコンテンツをエージェントが構造化された形式でアクセスするためのプロトコルです。従来のWebスクレイピングとは異なり、Webサイト側がエージェント向けの構造化データを提供する仕組みで、robots.txtに近い発想です。

Durable Executionでエージェントの耐障害性を実現する

AIエージェントの本番運用で最も厄介な問題の1つが、長時間実行タスクの途中障害です。例えば、エージェントが10ステップのワークフローの7ステップ目でAPIタイムアウトを起こした場合、従来のアプローチではワークフロー全体をやり直すか、複雑なチェックポイント処理を自前で実装する必要がありました。

Durable Execution は、この問題を「インフラの責務」として解決するアプローチです。ワークフローの各ステップの状態を自動的に永続化し、障害発生時に中断箇所から自動再開します。

Temporalで耐障害性のあるエージェントワークフローを実装する

Temporalは2026年2月にシリーズDで$300Mを調達し、$5Bの評価額を得ました。OpenAI、ADP、Yum! Brands、Blockなどが本番環境で利用しており、OpenAIのアプリケーションインフラVPは「Durable Executionは現代のAIシステムの中核要件」と述べています（Temporal公式ブログ）。

# workflows.py - Temporal + Pydantic AIでDurable AIエージェントを実装する例
# Python 3.11+, pip install temporalio pydantic-ai
from datetime import timedelta
from dataclasses import dataclass
from temporalio import workflow, activity
from temporalio.common import RetryPolicy

@dataclass
class AgentTaskInput:
    user_query: str
    context: dict
    max_retries: int = 3

@dataclass
class AgentTaskResult:
    response: str
    tools_used: list[str]
    total_tokens: int

@activity.defn
async def call_llm_with_tools(task: AgentTaskInput) -> AgentTaskResult:
    """LLMにツール付きでリクエストを送信するアクティビティ

    Temporal が自動リトライ・タイムアウト管理を行うため、
    アクティビティ内のエラーハンドリングは最小限でよい。
    """
    from pydantic_ai import Agent

    agent = Agent(
        "anthropic:claude-sonnet-4-20250514",
        system_prompt="あなたはデータ分析アシスタントです。",
    )
    result = await agent.run(task.user_query)

    return AgentTaskResult(
        response=result.output,
        tools_used=["database_search", "chart_generation"],
        total_tokens=result.usage().total_tokens,
    )

@activity.defn
async def save_result_to_db(result: AgentTaskResult) -> str:
    """結果をデータベースに保存するアクティビティ"""
    # DB保存ロジック（省略）
    return "saved"

@workflow.defn
class AgentWorkflow:
    """Durable Executionで保護されたAIエージェントワークフロー

    各アクティビティ（LLM呼び出し、DB保存）が自動的に永続化され、
    障害発生時は中断箇所から再開する。
    """

    @workflow.run
    async def run(self, task: AgentTaskInput) -> AgentTaskResult:
        # ステップ1: LLMにツール付きでリクエスト
        # タイムアウト: 60秒、リトライ: 最大3回（指数バックオフ）
        result = await workflow.execute_activity(
            call_llm_with_tools,
            task,
            start_to_close_timeout=timedelta(seconds=60),
            retry_policy=RetryPolicy(
                initial_interval=timedelta(seconds=1),
                maximum_interval=timedelta(seconds=30),
                maximum_attempts=task.max_retries,
            ),
        )

        # ステップ2: 結果をDB保存
        # ← ステップ1が成功した状態が永続化されているため、
        #    ここで障害が起きてもステップ1はやり直さない
        await workflow.execute_activity(
            save_result_to_db,
            result,
            start_to_close_timeout=timedelta(seconds=10),
        )

        return result

なぜTemporalを選ぶのか:

• 自動状態永続化: ワークフローの各ステップ完了時に状態がイベントソーシングで記録される
• 宣言的リトライ: RetryPolicyを宣言するだけで指数バックオフ+ジッタが実装される
• 長時間実行対応: 数時間〜数日間のワークフローもサポート（エージェントが人間の承認を待つケースなど）

制約条件: Temporal Cloudの料金体系はアクション数に基づくため、エージェントが大量のツール呼び出しを行うワークフローではコストが増大する可能性があります。小規模なプロジェクトではセルフホストのTemporal Server、またはRestateのような軽量代替を検討してください。

Restateとの比較: 軽量Durable Execution

Temporalの代替として注目されているのが Restate です。$7Mのシード資金を調達し、AWS LambdaやCloudflare Workersとネイティブに統合する軽量Durable Executionプラットフォームです。

観点	Temporal	Restate
アーキテクチャ	サーバー + ワーカークラスター	軽量シングルバイナリ
状態管理	イベントソーシング	Virtual Objects
サーバーレス対応	Temporal Cloud（マネージド）	Lambda / Workers ネイティブ統合
レイテンシ	ミリ秒〜秒（ネットワーク依存）	超低レイテンシ（FaaS最適化）
適用シーン	大規模エンタープライズ、複雑なワークフロー	サーバーレス、低レイテンシ要件
資金規模	$300M（シリーズD、評価額$5B）	$7M（シード）

ハマりポイント: Restateはサーバーレス環境でのレイテンシに優れますが、Temporalほどのエコシステム（SDK、UIツール、コミュニティ）は成熟していません。本番環境でのサポート体制を重視するならTemporal、開発速度とサーバーレスとの親和性を重視するならRestateという判断になります。

エージェントSDKを比較し適切なプラットフォームを選定する

2026年3月時点で、AIエージェント構築のための主要なSDK/フレームワークは以下の通りです。実際に試してみると、各SDKの設計思想の違いが明確に見えてきます。

主要SDKの比較

SDK/フレームワーク	提供元	言語	MCP対応	A2A対応	特徴
OpenAI Agents SDK	OpenAI	Python（Node.js予定）	Responses API経由	未対応	Swarmの後継、Responses API統合
Claude Agent SDK	Anthropic	Python	ネイティブ対応	未対応	旧Claude Code SDK、MCP発明元
Google ADK	Google	Python	ネイティブ対応	ネイティブ対応	ADK + A2A + MCPの3層統合
LangGraph	LangChain	Python / JS	ツール経由	未対応	循環グラフ、明示的状態マシン
CrewAI	CrewAI	Python	ツール経由	独自実装	ロールベース、協調型マルチエージェント

Google ADKでMCP + A2Aエージェントを実装する

Google ADKは、MCP・A2A・Geminiモデルを統合した開発キットです。3層プロトコルスタックすべてをネイティブにサポートする唯一のSDKです（Google ADK公式ドキュメント）。

# google_adk_agent.py - Google ADKでMCPツール連携エージェントを構築する例
# Python 3.11+, pip install google-adk
from google.adk.agents import Agent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import StdioServerParameters

# MCPサーバーからツールを自動読み込み
mcp_tools = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="python",
            args=["mcp_server.py"],  # 先ほど作成したMCPサーバー
        )
    )
)

# エージェント定義
research_agent = Agent(
    name="research_agent",
    model="gemini-2.5-pro",
    description="リサーチと情報収集を担当するエージェント",
    instruction="ユーザーの質問に対して、利用可能なツールを活用して調査し回答してください。",
    tools=[mcp_tools],
)

なぜGoogle ADKか:

• MCP + A2Aの両方をネイティブサポートし、マルチエージェント構成への拡張が容易
• モデル非依存設計で、Gemini以外のLLMも利用可能

注意点:

Google ADKは2025年後半にGA（一般提供）されたばかりで、OpenAI/AnthropicのSDKと比べるとコミュニティの事例やサードパーティ統合がまだ少ない状況です。本番利用前にPoCで自社要件との適合性を検証することを推奨します。

OpenAI Agents SDKの実装例

OpenAI Agents SDKは、2025年のSwarmの後継として位置づけられ、Responses APIと統合されたエージェントフレームワークです（OpenAI公式ドキュメント）。

# openai_agent.py - OpenAI Agents SDKでの実装例
# Python 3.11+, pip install openai-agents
from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: str) -> str:
    """指定された都市の天気情報を取得する"""
    # 実際にはWeather APIを呼び出す
    return f"{city}の天気: 晴れ、気温22度"

@function_tool
def search_flights(origin: str, destination: str, date: str) -> str:
    """フライト情報を検索する"""
    return f"{origin} -> {destination} ({date}): 3件のフライトが見つかりました"

# エージェント定義
travel_agent = Agent(
    name="travel_assistant",
    instructions="あなたは旅行アシスタントです。天気やフライト情報を活用してユーザーの旅行計画を支援してください。",
    tools=[get_weather, search_flights],
    model="gpt-4o",
)

# 実行
async def main():
    result = await Runner.run(travel_agent, "来週の東京から大阪への出張を計画して")
    print(result.final_output)

注意: OpenAIのAssistants APIは2026年8月にサンセット予定です（OpenAI公式）。新規開発ではResponses API + Agents SDKを使用してください。

エージェントオブザーバビリティを構築する

AIエージェントが本番環境で動作し始めると、「なぜこのエージェントはこの応答を返したのか」「どのツール呼び出しでボトルネックが発生しているのか」を追跡する必要があります。従来のAPM（Application Performance Monitoring）では、LLMの推論ステップやツール呼び出しチェーンの可視化が困難です。

2026年時点のAIエージェントオブザーバビリティは、LangSmith（商用）と Langfuse（OSS）の2つが主要な選択肢です（AIMultiple 2026調査）。

LangSmithとLangfuseの比較

観点	LangSmith	Langfuse
ライセンス	商用（無料枠あり）	MIT（オープンソース）
GitHubスター	N/A（商用）	19,000+
トレーシングオーバーヘッド	ほぼゼロ	約15%
セルフホスト	不可	可能
プロンプト管理	あり	あり（バージョニング対応）
評価機能	自動評価 + ヒューマンレビュー	フレキシブル評価パイプライン
フレームワーク対応	LangChain優先、他SDKも対応	フレームワーク非依存
推奨シーン	本番パフォーマンス重視	セルフホスト・カスタマイズ重視

Langfuseでエージェントトレーシングを実装する

# observability.py - Langfuseでエージェントのトレーシングを実装する例
# Python 3.11+, pip install langfuse openai
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
import openai

# Langfuseクライアント初期化
# 環境変数: LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_HOST
langfuse = Langfuse()

@observe(as_type="generation")
def call_llm(messages: list[dict], model: str = "gpt-4o") -> str:
    """LLM呼び出しをLangfuseでトレースする"""
    client = openai.OpenAI()
    response = client.chat.completions.create(
        model=model,
        messages=messages,
    )

    # トークン使用量をLangfuseに記録
    langfuse_context.update_current_observation(
        usage={
            "input": response.usage.prompt_tokens,
            "output": response.usage.completion_tokens,
        },
        model=model,
    )
    return response.choices[0].message.content

@observe()
def search_tool(query: str) -> list[dict]:
    """ツール呼び出しもトレースする"""
    # DB検索などの処理
    langfuse_context.update_current_observation(
        metadata={"query": query, "result_count": 5}
    )
    return [{"title": "結果1", "score": 0.95}]

@observe()
def agent_pipeline(user_query: str) -> str:
    """エージェントパイプライン全体をトレースする

    observe()デコレータのネストにより、
    パイプライン > LLM呼び出し > ツール呼び出し の階層がトレースされる。
    """
    # ステップ1: ツール検索
    search_results = search_tool(user_query)

    # ステップ2: LLM呼び出し（検索結果をコンテキストに含む）
    messages = [
        {"role": "system", "content": "検索結果を基に回答してください。"},
        {"role": "user", "content": f"質問: {user_query}\n検索結果: {search_results}"},
    ]
    response = call_llm(messages)

    return response

なぜオブザーバビリティが必要か:

• エージェントの推論ステップが不透明なため、「なぜこの回答になったか」を追跡する手段が必要
• ツール呼び出しのレイテンシ・エラー率を監視し、ボトルネックを特定する
• トークン使用量・コストを可視化し、最適化のデータを得る

トレードオフ: Langfuseはセルフホスト可能でカスタマイズ性が高い反面、トレーシングオーバーヘッドが約15%と報告されています。レイテンシに厳しい要件がある場合はLangSmithを検討してください。

OpenTelemetryとの統合

2026年のAIオブザーバビリティ業界は、OpenTelemetry（OTEL） をエージェントテレメトリの収集標準として収束しつつあります。New Relicが2026年2月にリリースしたAIエージェントプラットフォームもOpenTelemetryベースです（TechCrunch報道）。

Cloudflare Workers + Durable Objectsでサーバーレスエージェントを構築する

サーバーを自前で管理したくない場合、Cloudflareが提供するサーバーレスAIエージェント基盤が有力な選択肢です。Durable Objectsでエージェントの状態を永続化し、Workers上でMCPサーバーをホストできます（Cloudflare公式ブログ）。

// agent.ts - Cloudflare Workers + Agents SDKでMCPエージェントを構築する例
// TypeScript 5.x, wrangler 4.x
import { Agent, routeAgentRequest } from "agents";

interface AgentState {
  conversationHistory: Array<{ role: string; content: string }>;
  preferences: Record<string, string>;
}

export class MyAgent extends Agent<Env, AgentState> {
  // 初期状態の定義
  initialState: AgentState = {
    conversationHistory: [],
    preferences: {},
  };

  async onChatMessage(message: string): Promise<string> {
    // 会話履歴に追加（Durable Objectが自動永続化）
    this.setState({
      ...this.state,
      conversationHistory: [
        ...this.state.conversationHistory,
        { role: "user", content: message },
      ],
    });

    // LLMに問い合わせ（MCPツールは自動検出）
    const response = await this.env.AI.run("@cf/meta/llama-3.3-70b-instruct-fp8-fast", {
      messages: this.state.conversationHistory,
    });

    // 応答を履歴に追加
    this.setState({
      ...this.state,
      conversationHistory: [
        ...this.state.conversationHistory,
        { role: "assistant", content: response.response },
      ],
    });

    return response.response;
  }
}

// Agents SDKのルーティングを使用
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    return routeAgentRequest(request, env);
  },
};

Cloudflareアプローチの利点:

• エージェントはアイドル時にハイバネートし、リクエスト時に状態を復元して起動するため、コストが最小限
• Durable Objects Free Tierが提供されており、小規模なエージェントは無料で運用可能
• Remote MCPサーバーのホスティングに最適で、インターネット経由でエージェントが接続可能

制約条件: Cloudflare Workers の実行時間制限（CPU時間30秒、wall clock 15分）があるため、大規模なデータ処理や長時間のLLM推論には不向きです。そのようなケースでは、Temporal + 専用コンピュートインスタンスの組み合わせが適しています。

よくある問題と解決方法

問題	原因	解決方法
MCPサーバーに接続できない	stdio/HTTP のトランスポート設定不一致	サーバーとクライアントのトランスポート種別を統一する
エージェントがツールを検出しない	MCPサーバーのtool定義にdocstringがない	@mcp.tool() デコレータの関数にdocstringを必ず追加する
Temporalワークフローがタイムアウト	start_to_close_timeoutが短すぎる	LLM応答時間を考慮して60秒以上に設定する
Langfuseにトレースが表示されない	環境変数（APIキー）の設定漏れ	LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_HOST を確認する
A2Aエージェント間で認証エラー	Agent Cardの認証設定不備	A2AのAgent Card JSONに正しい認証スキーマを記載する
Durable Objectsの状態が消える	ハイバネーション後の復元処理未実装	setState で状態を明示的に永続化する

まとめと次のステップ

まとめ:

• AIエージェント基盤は MCP（ツール接続）・A2A（エージェント間通信）・WebMCP（Web構造化アクセス）の3層プロトコルスタックで構成される
• Durable Execution（Temporal / Restate）がエージェントの耐障害性を「インフラの責務」として解決し、手動リトライコードを不要にする
• エージェントSDKは用途で選択: MCPネイティブならClaude Agent SDK、A2A統合ならGoogle ADK、OpenAIエコシステムならAgents SDK
• オブザーバビリティ はLangSmith（低オーバーヘッド・商用）またはLangfuse（OSS・セルフホスト）で構築し、OpenTelemetryとの統合が業界標準になりつつある
• サーバーレスエージェントにはCloudflare Workers + Durable Objectsが有力な選択肢

次にやるべきこと:

• MCP公式のQuickstartでMCPサーバーを1つ構築してみる
• Temporal公式チュートリアルでDurable Executionの基本を体験する
• Langfuse Self-hostingガイドでオブザーバビリティ環境を構築する

参考

• Temporal公式ブログ: $300M Series D at $5B Valuation
• Google ADK公式ドキュメント: A2A Protocol
• Cloudflare: Building AI Agents with MCP, Auth, and Durable Objects
• OpenAI Agents SDK公式ドキュメント
• Anthropic: Building Agents with the Claude Agent SDK
• AIMultiple: 15 AI Agent Observability Tools in 2026
• The 2026 AI Agent Protocol Wars Explained
• 日経クロステック: グランプリはMCP、AIエージェントと業務システムをつなぐ共通基盤に
• TechCrunch: New Relic launches new AI agent platform and OpenTelemetry tools
• Langfuse公式: AI Agent Observability

---

注意: この記事はAI（Claude Code）により自動生成されました。内容の正確性については複数の情報源で検証していますが、実際の利用時は公式ドキュメントもご確認ください。