附录 B：代码示例集

B.1 MCP Server 连接示例

# MCP Discovery + Configuration + Connection
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# Step 1: Discovery（从 Registry 找到 server）
server_params = StdioServerParameters(
    command="python",
    args=["mcp_server_bigquery.py"],
    env={"GOOGLE_APPLICATION_CREDENTIALS": "/path/to/key.json"}
)

# Step 2: Connection
async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        # Step 3: Initialize
        await session.initialize()
        
        # List available tools
        tools = await session.list_tools()
        for tool in tools.tools:
            print(f"Tool: {tool.name}, Description: {tool.description}")
        
        # Call a tool
        result = await session.call_tool("query_bigquery", arguments={
            "project_id": "my-project",
            "query": "SELECT * FROM dataset.table LIMIT 10"
        })
        print(result.content)

B.2 A2A Agent 暴露示例

# Supply-side: Exposing an A2A Agent
from google.adk.agents import LlmAgent
from google.adk.models import Gemini
from google.adk.runners import A2aAgentExecutor, A2aAgentExecutorConfig

# Define Agent Card
agent_card = {
    "name": "billing-specialist",
    "description": "Handles billing inquiries and payment processing",
    "capabilities": ["query_billing", "process_payment", "generate_invoice"],
    "security": {"data_handling": "encrypted", "auth_required": True}
}

# Create Agent
billing_agent = LlmAgent(
    name="billing_agent",
    model=Gemini(model="gemini-flash-latest"),
    instruction="You are a billing specialist...",
    tools=[query_billing_tool, process_payment_tool]
)

# Expose as A2A Endpoint
executor_config = A2aAgentExecutorConfig(
    agent_card=agent_card,
    endpoint="/a2a/billing"
)

executor = A2aAgentExecutor(billing_agent, executor_config)
# Deploy executor to cloud endpoint

B.3 A2A Remote Agent 连接示例

# Demand-side: Connecting to Remote A2A Agents
from google.adk.agents import RemoteA2aAgent, AgentRegistry

# Option 1: Direct Instantiation
billing_specialist = RemoteA2aAgent(
    name="billing_agent",
    endpoint="https://api.vendor.com/v1/billing/a2a"
)

# Option 2: Registry Discovery
registry = AgentRegistry(project_id="my-project", location="us-central1")
agent_name = f"projects/my-project/locations/us-central1/agents/billing-agent-id"
billing_specialist = registry.get_remote_a2a_agent(agent_name=agent_name)

# Use in Orchestrator
orchestrator = LlmAgent(
    name="orchestrator",
    model=Gemini(model="gemini-pro-latest"),
    instruction="You are an orchestrator...",
    sub_agents=[billing_specialist, hr_specialist, compliance_specialist]
)

B.4 A2UI Tool-as-template 示例

from google.adk.agents import LlmAgent
from google.adk.models import Gemini
from a2ui.adk.send_a2ui_to_client_toolset import A2uiPartConverter

def get_sales_dashboard(region: str) -> dict:
    """Build a data-bound sales dashboard for `region`."""
    data = fetch_sales(region)
    return {
        "version": "v0.9",
        "updateComponents": {
            "surfaceId": "sales",
            "components": [
                {"id": "root", "component": "Column", "children": ["title", "total", "drill"]},
                {"id": "title", "component": "Text", "text": {"path": "/title"}, "variant": "h1"},
                {"id": "total", "component": "Text", "text": {"path": "/total"}},
                {"id": "drill", "component": "Button", "child": "drill-label", "action": {"event": {"name": "expand_details"}}},
                {"id": "drill-label", "component": "Text", "text": "Drill Down"}
            ]
        }
    }

agent = LlmAgent(
    name="sales_agent",
    model=Gemini(model="gemini-flash-latest"),
    tools=[get_sales_dashboard]
)

# Wire converter to make tool response an A2UI part
executor_config = A2aAgentExecutorConfig(
    event_converter=A2uiPartConverter(catalog, bypass_tool_check=True)
)

B.5 AP2/UCP 流程示例（概念）

┌─────────────────────────────────────────────────────────────┐
│  UCP 流程                                                    │
├─────────────────────────────────────────────────────────────┤
│  ① Agent 查询商品 catalog                                   │
│     UCP request: {product_id: "burrito-001", options: ["veg"]}│
│ ② Merchant 响应价格、税费、配送费                           │
│     UCP response: {price: 15.00, tax: 1.50, delivery: 2.00}│
│ ③ Agent 构建订单                                            │
│     UCP order: {items: [...], total: 18.50}                 │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│  AP2 流程                                                    │
├─────────────────────────────────────────────────────────────┤
│  ① 用户批准 Mandate："可在 Taco Bell 消费最多 $25"          │
│ ② Agent 生成加密 promissory note                            │
│     AP2 token: signed("human approved $18.50 order")       │
│ ③ Merchant 验证签名                                         │
│ ④ Payment processor 执行交易                                │
│ ⑤ AP2 审计日志记录                                          │
└─────────────────────────────────────────────────────────────┘

---