MCP と A2A

// 2026-07-28 の tools/call リクエスト例（ハンドシェイク不要）
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "search",
    "arguments": {"q": "MCP stateless"},
    "_meta": {
      "io.modelcontextprotocol/protocolVersion": "2026-07-28",
      "io.modelcontextprotocol/clientInfo": {"name": "my-app", "version": "1.0"}
    }
  }
}

User → Client Agent → [A2A JSON-RPC over HTTP] → Remote Agent (Server)
                                                    │
                                           Agent Card で発見
                                           (/.well-known/agent-card.json)

submitted → working → completed
                  ↘ input-required → 再開 → completed
                  ↘ failed
                  ↘ canceled

{
  "name": "Code Review Agent",
  "description": "コードのセキュリティ・パフォーマンス・スタイルをレビュー",
  "url": "https://code-review.example.com/a2a",
  "version": "1.0.0",
  "capabilities": { "streaming": true, "pushNotifications": true },
  "authentication": { "schemes": ["Bearer"] },
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain"],
  "skills": [
    {
      "id": "security-scan",
      "name": "セキュリティスキャン",
      "description": "OWASP Top 10 の脆弱性を検出",
      "tags": ["security", "code-review"]
    },
    {
      "id": "performance-analysis",
      "name": "パフォーマンス分析",
      "description": "N+1クエリ、メモリリーク、非効率なアルゴリズムを検出",
      "tags": ["performance", "optimization"]
    }
  ]
}

Transport（通信）→ Gate（認証）→ Box（コンテナ化）→ Edge（リバースプロキシ+TLS）

import os

MCP_TOKEN = os.environ.get("MCP_TOKEN", "")

app = MCPServer("notes-server")

# v2 の認証機構を利用。または Nginx 側で認証を肩代わりする（Step 4 で後述）

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY pyproject.toml .
RUN pip install --no-cache-dir mcp aiosqlite uvicorn
COPY server.py .

EXPOSE 8000
CMD ["python", "server.py"]

# /etc/nginx/sites-available/mcp
server {
    listen 443 ssl http2;
    server_name mcp.example.com;

    ssl_certificate /etc/letsencrypt/live/mcp.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/mcp.example.com/privkey.pem;

    location /mcp {
        proxy_pass http://127.0.0.1:8000;
        proxy_http_version 1.1;
        proxy_set_header Connection "";

        # ★ 重要: SSE ストリーミングのための設定
        proxy_buffering off;
        proxy_cache off;
        chunked_transfer_encoding on;
        proxy_read_timeout 300s;

        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

server {
    listen 80;
    server_name mcp.example.com;
    return 301 https://$host$request_uri;
}

# /etc/systemd/system/mcp-server.service
[Unit]
Description=MCP Notes Server
After=network-online.target docker.service
Requires=docker.service

[Service]
Type=oneshot
RemainAfterExit=yes
ExecStart=/usr/bin/docker compose -f /opt/mcp-server/docker-compose.yml up -d
ExecStop=/usr/bin/docker compose -f /opt/mcp-server/docker-compose.yml down
Restart=on-failure
RestartSec=10

[Install]
WantedBy=multi-user.target

# agent_card.py
from a2a.types import (
    AgentCapabilities, AgentCard, AgentInterface, AgentSkill,
)

skill = AgentSkill(
    id="reverse-text",
    name="Reverse Text",
    description="受け取ったテキストを反転して返す",
    tags=["text", "utility"],
    examples=["hello", "world"],
)

public_agent_card = AgentCard(
    name="Reverse Agent",
    description="テキスト反転エージェント",
    version="1.0.0",
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    capabilities=AgentCapabilities(streaming=True),
    supported_interfaces=[
        AgentInterface(
            protocol_binding="JSONRPC",
            url="http://127.0.0.1:9999",
            protocol_version="1.0",
        )
    ],
    skills=[skill],
)

# agent_executor.py
import asyncio
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.types import Task, TaskState, TaskStatus, TaskStatusUpdateEvent
from a2a.utils.message import new_agent_text_message


class ReverseAgentExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue):
        user_text = context.message.parts[0].root.text

        await event_queue.put(
            TaskStatusUpdateEvent(
                task_id=context.task_id,
                context_id=context.context_id,
                status=TaskStatus(state=TaskState.working),
                final=False,
            )
        )

        reversed_text = user_text[::-1]

        await event_queue.put(
            Task(
                id=context.task_id,
                context_id=context.context_id,
                status=TaskStatus(
                    state=TaskState.completed,
                    message=new_agent_text_message(
                        reversed_text, context.context_id, context.task_id,
                    ),
                ),
            )
        )

    async def cancel(self, context: RequestContext, event_queue: EventQueue):
        await event_queue.put(
            Task(
                id=context.task_id,
                context_id=context.context_id,
                status=TaskStatus(state=TaskState.canceled),
            )
        )

python server.py

# Agent Card の取得
curl -s http://127.0.0.1:9999/.well-known/agent-card.json | python -m json.tool

# テキスト反転の依頼（A2A-Version ヘッダ必須）
curl -s -X POST http://127.0.0.1:9999/ \
  -H "Content-Type: application/json" \
  -H "A2A-Version: 1.0" \
  -d '{
    "jsonrpc": "2.0",
    "id": "req-1",
    "method": "SendMessage",
    "params": {
      "message": {
        "role": "user",
        "parts": [{"kind": "text", "text": "Hello A2A!"}],
        "messageId": "msg-1"
      }
    }
  }'

# client.py
import asyncio
from a2a.client import A2AClient


async def main():
    async with A2AClient(url="http://localhost:9999") as client:
        card = await client.get_agent_card()
        print(f"Connected to: {card.name}")
        print(f"Skills: {[s.name for s in card.skills]}")

        response = await client.send_message(
            message={
                "role": "user",
                "parts": [{"kind": "text", "text": "Protocol tutorial"}],
            }
        )
        task = response.result
        reply = task.status.message.parts[0].root.text
        print(f"Agent replied: {reply}")


asyncio.run(main())

from google.adk import Agent
from google.adk.a2a.utils.agent_to_a2a import to_a2a

agent = Agent(
    model="gemini-2.0-flash",
    name="smart_reverse",
    description="受け取ったテキストを反転して返す",
    instruction="ユーザーから受け取ったテキストを反転して返してください。",
)

app = to_a2a(agent, port=8001)
# uvicorn adk_agent:app --host localhost --port 8001

ユーザー → Orchestrator Agent
              │ A2A（タスク委譲）
              ├── Research Agent ── MCP（Web検索・スクレイピング）
              ├── Compliance Agent ── MCP（法令DB・契約書DB）
              └── Report Agent ── MCP（CMSツール）