本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

# 在 AgentCore 執行期中部署 A2A 伺服器
<a name="runtime-a2a"></a>

Amazon Bedrock AgentCore 執行期可讓您在 AgentCore 執行期中部署和執行 Agent-to-Agent (A2A) 伺服器。本指南會逐步引導您建立、測試和部署第一個 A2A 伺服器。

在本區段，您會學習：
+ Amazon Bedrock AgentCore 如何支援 A2A
+ 如何使用代理程式功能建立 A2A 伺服器
+ 如何在本機測試您的伺服器
+ 如何將伺服器部署至 AWS 
+ 如何叫用已部署的伺服器
+ 如何擷取代理程式卡以進行探索

如需 A2A 的詳細資訊，請參閱 [A2A 通訊協定合約](runtime-a2a-protocol-contract.md)。

**Topics**
+ [Amazon Bedrock AgentCore 如何支援 A2A](#runtime-a2a-how-agentcore-supports)
+ [搭配 AgentCore 執行期使用 A2A](#runtime-a2a-steps)
+ [附錄](#runtime-a2a-appendix)

## Amazon Bedrock AgentCore 如何支援 A2A
<a name="runtime-a2a-how-agentcore-supports"></a>

Amazon Bedrock AgentCore 的 A2A 通訊協定支援可做為透明代理層，與 A2A 伺服器無縫整合。為 A2A 設定時，Amazon Bedrock AgentCore 預期容器在根路徑 (`0.0.0.0:9000/`) `9000`的連接埠上執行無狀態、可串流的 HTTP 伺服器，這符合預設的 A2A 伺服器組態。

此服務提供企業級工作階段隔離，同時維持通訊協定透明度 - 來自 [InvokeAgentRuntime](https://docs.aws.amazon.com/bedrock-agentcore/latest/APIReference/API_InvokeAgentRuntime.html) API 的 JSON-RPC 承載會直接傳遞至 A2A 容器，無需修改。此架構會保留標準 A2A 通訊協定功能，例如透過 的代理程式卡`/.well-known/agent-card.json`和 JSON-RPC 通訊的內建代理程式探索，同時新增企業身分驗證 (SigV4/OAuth 2.0) 和可擴展性。

與其他通訊協定的主要區別在於連接埠 (9000 vs 8080 for HTTP)、掛載路徑 ( `/` vs `/invocations` ) 和標準化代理程式探索機制，讓 Amazon Bedrock AgentCore 成為生產環境中 A2A 代理程式的理想部署平台。

與其他通訊協定的主要差異：

 **連接埠**   
A2A 伺服器在連接埠 9000 上執行 （相較於 HTTP 的 8080、MCP 的 8000)

 **路徑**   
A2A 伺服器掛載於 `/`(vs `/invocations` for HTTP， `/mcp` for MCP)

 **客服人員卡**   
A2A 透過位於 的代理程式卡提供內建代理程式探索 `/.well-known/agent-card.json` 

 **通訊協定**   
使用 JSON-RPC agent-to-agent的通訊

 **身分驗證**   
同時支援 SigV4 和 OAuth 2.0 身分驗證機制

如需詳細資訊，請參閱[https://a2a-protocol.org/](https://a2a-protocol.org/)。

## 搭配 AgentCore 執行期使用 A2A
<a name="runtime-a2a-steps"></a>

在本教學課程中，您會建立、測試和部署 A2A 伺服器。

**Topics**
+ [先決條件](#runtime-a2a-prerequisites)
+ [步驟 1：建立您的 A2A 專案](#runtime-a2a-create-server)
+ [步驟 2：在本機測試您的 A2A 伺服器](#runtime-a2a-test-locally)
+ [步驟 3：將 A2A 伺服器部署至 Bedrock AgentCore 執行期](#runtime-a2a-deploy)
+ [步驟 4：取得客服人員卡](#runtime-a2a-step-4)
+ [步驟 5：叫用您部署的 A2A 伺服器](#runtime-a2a-step-5)

### 先決條件
<a name="runtime-a2a-prerequisites"></a>
+ 已安裝 Python 3.10 或更新版本並基本了解 Python
+ 已安裝 Node.js 20 或更高版本 (AgentCore CLI 需要）
+ 已安裝 AgentCore CLI： `npm install -g @aws/agentcore`
+ 已設定適當許可和本機登入資料的 AWS 帳戶
+ 了解 A2A 通訊協定和agent-to-agent的通訊概念

### 步驟 1：建立您的 A2A 專案
<a name="runtime-a2a-create-server"></a>

此範例使用 Strands Agents，但 AgentCore CLI 也支援具有 LangChain/LangGraph 和 Google ADK 的 A2A 專案。

#### Scaffold 專案
<a name="runtime-a2a-scaffold-project"></a>

執行以下命令：

```
agentcore create \
  --project-name A2AProject \
  --name A2AAgent \
  --language Python \
  --framework Strands \
  --model-provider Bedrock \
  --memory none \
  --protocol A2A
cd A2AProject
```

CLI 會使用所有必要的相依性和組態來堆疊完整的專案。產生的 `main.py`包含您的 A2A 伺服器：

```
from strands import Agent, tool
from strands.multiagent.a2a.executor import StrandsA2AExecutor
from bedrock_agentcore.runtime import serve_a2a
from model.load import load_model

@tool
def add_numbers(a: int, b: int) -> int:
    """Return the sum of two numbers."""
    return a + b

tools = [add_numbers]

agent = Agent(
    model=load_model(),
    system_prompt="You are a helpful assistant. Use tools when appropriate.",
    tools=tools,
)

if __name__ == "__main__":
    serve_a2a(StrandsA2AExecutor(agent))
```

#### 了解程式碼
<a name="runtime-a2a-understanding-code"></a>

 **Strands 代理程式**   
建立具有特定工具和功能的代理程式

 **StrandsA2AExecutor**   
包裝 Strands 代理程式以提供 A2A 通訊協定相容性

 **serve\_a2a**   
啟動 Bedrock 相容 A2A 伺服器的 Amazon Bedrock AgentCore SDK 協助程式。根據預設，它會處理`/ping`運作狀態端點、代理程式卡服務、`AGENTCORE_RUNTIME_URL`環境變數、Batrock 標頭傳播，並在連接埠 9000 上執行。

 **連接埠 9000**   
根據預設，A2A 伺服器會在 AgentCore 執行時間中的連接埠 9000 上執行

若要自訂此代理程式，請將`add_numbers`工具取代為您自己的工具，並更新系統提示。

### 步驟 2：在本機測試您的 A2A 伺服器
<a name="runtime-a2a-test-locally"></a>

在本機開發環境中執行和測試您的 A2A 伺服器。

#### 啟動您的 A2A 伺服器
<a name="runtime-a2a-start-server"></a>

使用 AgentCore CLI 在本機啟動 A2A 伺服器：

```
agentcore dev
```

這會在 Web 瀏覽器中開啟 AgentCore 代理程式檢查程式。若要改為使用終端機型 TUI，請使用 `agentcore dev --no-browser`。

或者，您可以直接執行伺服器：

```
python main.py
```

您應該會看到輸出，指出伺服器正在連接埠 上執行`9000`。

#### 叫用代理程式
<a name="runtime-a2a-invoke-agent"></a>

```
curl -X POST http://localhost:9000/ \
-H "Content-Type: application/json" \
-d '{
  "jsonrpc": "2.0",
  "id": "req-001",
  "method": "message/send",
  "params": {
    "message": {
      "role": "user",
      "parts": [
        {
          "kind": "text",
          "text": "what is 101 * 11?"
        }
      ],
      "messageId": "12345678-1234-1234-1234-123456789012"
    }
  }
}' | jq .
```

#### 測試代理程式卡擷取
<a name="runtime-a2a-test-agent-card"></a>

您可以在本機測試代理程式卡端點：

```
curl http://localhost:9000/.well-known/agent-card.json | jq.
```

您也可以使用 A2A Inspector 測試已部署的伺服器，如[使用 A2A 檢查器的遠端測試](https://github.com/a2aproject/a2a-inspector)中所述。

### 步驟 3：將 A2A 伺服器部署至 Bedrock AgentCore 執行期
<a name="runtime-a2a-deploy"></a>

#### 設定 Cognito 使用者集區以進行身分驗證
<a name="runtime-a2a-setup-cognito"></a>

部署之前，請設定身分驗證以安全存取已部署的伺服器。如需 Cognito 設定說明的詳細資訊，請參閱[設定 Cognito 使用者集區以進行身分驗證](runtime-mcp.md#runtime-mcp-appendix-a)。這可提供安全存取已部署伺服器所需的 OAuth 權杖。

#### 部署至 AWS
<a name="runtime-a2a-deploy-aws"></a>

部署您的代理程式：

```
agentcore deploy
```

此命令將：

1. 封裝您的代理程式程式碼和相依性

1. 將部署成品上傳至 Amazon S3

1. 建立 Amazon Bedrock AgentCore 執行期

1. 將您的代理程式部署到 AWS 

部署之後，您會收到客服人員執行期 ARN，如下所示：

```
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_a2a_server-xyz123
```

### 步驟 4：取得客服人員卡
<a name="runtime-a2a-step-4"></a>

代理程式卡是描述 A2A 伺服器身分、功能、技能、服務端點和身分驗證要求的 JSON 中繼資料文件。它們可在 A2A 生態系統中啟用自動代理程式探索。

#### 設定環境變數
<a name="runtime-a2a-step-4-setup-environment-variables"></a>

設定環境變數

1. 匯出承載字符做為環境變數。如需承載字符設定，請參閱[承載字符設定](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-mcp.html#runtime-mcp-appendix)。

   ```
   export BEARER_TOKEN="<BEARER_TOKEN>"
   ```

1. 匯出代理程式 ARN。

   ```
   export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_a2a_server-xyz123"
   ```

#### 擷取客服人員卡
<a name="retrieve-agent-card"></a>

```
import os
import json
import requests
from uuid import uuid4
from urllib.parse import quote

def fetch_agent_card():
    # Get environment variables
    agent_arn = os.environ.get('AGENT_ARN')
    bearer_token = os.environ.get('BEARER_TOKEN')

    if not agent_arn:
        print("Error: AGENT_ARN environment variable not set")
        return

    if not bearer_token:
        print("Error: BEARER_TOKEN environment variable not set")
        return

    # URL encode the agent ARN
    escaped_agent_arn = quote(agent_arn, safe='')

    # Construct the URL
    url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{escaped_agent_arn}/invocations/.well-known/agent-card.json"

    # Generate a unique session ID
    session_id = str(uuid4())
    print(f"Generated session ID: {session_id}")

    # Set headers
    headers = {
        'Accept': '*/*',
        'Authorization': f'Bearer {bearer_token}',
        'X-Amzn-Bedrock-AgentCore-Runtime-Session-Id': session_id
    }

    try:
        # Make the request
        response = requests.get(url, headers=headers)
        response.raise_for_status()

        # Parse and pretty print JSON
        agent_card = response.json()
        print(json.dumps(agent_card, indent=2))

        return agent_card

    except requests.exceptions.RequestException as e:
        print(f"Error fetching agent card: {e}")
        return None

if __name__ == "__main__":
    fetch_agent_card()
```

從客服人員卡取得 URL 後，請將 匯出`AGENTCORE_RUNTIME_URL`為環境變數：

```
export AGENTCORE_RUNTIME_URL="https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/<ARN>/invocations/"
```

### 步驟 5：叫用您部署的 A2A 伺服器
<a name="runtime-a2a-step-5"></a>

建立用戶端程式碼來叫用您部署的 Amazon Bedrock AgentCore A2A 伺服器，並傳送訊息以測試功能。

建立新的檔案`my_a2a_client_remote.py`以叫用您部署的 A2A 伺服器：

```
import asyncio
import logging
import os
from uuid import uuid4

import httpx
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.types import Message, Part, Role, TextPart

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

DEFAULT_TIMEOUT = 300  # set request timeout to 5 minutes

def create_message(*, role: Role = Role.user, text: str) -> Message:
    return Message(
        kind="message",
        role=role,
        parts=[Part(TextPart(kind="text", text=text))],
        message_id=uuid4().hex,
    )

async def send_sync_message(message: str):
    # Get runtime URL from environment variable
    runtime_url = os.environ.get('AGENTCORE_RUNTIME_URL')

    # Generate a unique session ID
    session_id = str(uuid4())
    print(f"Generated session ID: {session_id}")

    # Add authentication headers for Amazon Bedrock AgentCore
    headers = {"Authorization": f"Bearer {os.environ.get('BEARER_TOKEN')}",
        'X-Amzn-Bedrock-AgentCore-Runtime-Session-Id': session_id}

    async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, headers=headers) as httpx_client:
        # Get agent card from the runtime URL
        resolver = A2ACardResolver(httpx_client=httpx_client, base_url=runtime_url)
        agent_card = await resolver.get_agent_card()

        # Agent card contains the correct URL (same as runtime_url in this case)
        # No manual override needed - this is the path-based mounting pattern

        # Create client using factory
        config = ClientConfig(
            httpx_client=httpx_client,
            streaming=False,  # Use non-streaming mode for sync response
        )
        factory = ClientFactory(config)
        client = factory.create(agent_card)

        # Create and send message
        msg = create_message(text=message)

        # With streaming=False, this will yield exactly one result
        async for event in client.send_message(msg):
            if isinstance(event, Message):
                logger.info(event.model_dump_json(exclude_none=True, indent=2))
                return event
            elif isinstance(event, tuple) and len(event) == 2:
                # (Task, UpdateEvent) tuple
                task, update_event = event
                logger.info(f"Task: {task.model_dump_json(exclude_none=True, indent=2)}")
                if update_event:
                    logger.info(f"Update: {update_event.model_dump_json(exclude_none=True, indent=2)}")
                return task
            else:
                # Fallback for other response types
                logger.info(f"Response: {str(event)}")
                return event

# Usage - Uses AGENTCORE_RUNTIME_URL environment variable
asyncio.run(send_sync_message("what is 101 * 11"))
```

## 附錄
<a name="runtime-a2a-appendix"></a>

**Topics**
+ [設定 Cognito 使用者集區以進行身分驗證](#runtime-a2a-setup-cognito-appendix)
+ [使用 A2A 檢查器進行遠端測試](#runtime-a2a-remote-testing)
+ [疑難排解](#runtime-a2a-troubleshooting)

### 設定 Cognito 使用者集區以進行身分驗證
<a name="runtime-a2a-setup-cognito-appendix"></a>

如需 Cognito 設定說明的詳細資訊，請參閱 MCP 文件中的設定 [Cognito 使用者集區以進行身分驗證](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-mcp.html#set-up-cognito-user-pool-for-authentication)。

### 使用 A2A 檢查器進行遠端測試
<a name="runtime-a2a-remote-testing"></a>

請參閱 [https://github.com/a2aproject/a2a-inspector](https://github.com/a2aproject/a2a-inspector)。

### 疑難排解
<a name="runtime-a2a-troubleshooting"></a>

 **常見 A2A-specific問題** 

以下是您可能遇到的常見問題：

連接埠衝突  
A2A 伺服器必須在 AgentCore 執行期環境中的連接埠 9000 上執行

JSON-RPC 錯誤  
檢查您的用戶端是否傳送格式正確的 JSON-RPC 2.0 訊息

授權方法不符  
確保您的請求使用與客服人員設定相同的身分驗證方法 (OAuth 或 SigV4)

 **例外狀況處理** 

錯誤處理的 A2A 規格： [https://a2a-protocol.org/latest/specification/#81-standard-json-rpc-errors](https://a2a-protocol.org/latest/specification/#81-standard-json-rpc-errors)

A2A 伺服器會以標準 JSON-RPC 錯誤回應的形式傳回大部分錯誤。服務會以原生 HTTP 錯誤的形式傳回身分驗證和授權失敗 （例如 `AccessDeniedException`) 及其自己的狀態碼，如下表所示。服務會自動將內部執行時間錯誤轉譯為 JSON-RPC 內部錯誤，以維持通訊協定合規。

此服務使用標準化 JSON-RPC 錯誤代碼提供 A2A-compliant錯誤回應：


| JSON-RPC 錯誤代碼 | 執行時間例外狀況 | HTTP 錯誤代碼 | JSON-RPC 錯誤訊息 | 
| --- | --- | --- | --- | 
| 不適用 |  `AccessDeniedException`  | 403 | 存取遭拒 （以標準 HTTP 錯誤傳回，而非 JSON-RPC 錯誤） | 
| -32051 |  `ResourceNotFoundException`  | 404 | 找不到資源 – 請求的資源不存在 | 
| -32052 |  `ValidationException`  | 400 | 驗證錯誤 – 無效請求資料 | 
| -32053 |  `ThrottlingException`  | 429 | 超過速率限制 – 請求過多 | 
| -32053 |  `ServiceQuotaExceededException`  | 429 | 超過速率限制 – 請求過多 | 
| -32054 |  `ConflictException`  | 409 | 資源衝突 – 資源已存在 | 
| -32054 |  `RetryableConflictException`  | 409 | 工作階段操作進行中，請重試 | 
| -32055 |  `RuntimeClientError`  | 424 | 執行期用戶端錯誤 – 如需詳細資訊，請檢查您的 CloudWatch 日誌。 | 
| -32603 |  `Any other exception`  | 500 | 內部錯誤 - 處理請求時發生意外錯誤 |