# 用 AIMock 讓 AI 智能體測試變得決定性 — 在單一連接埠模擬 LLM、向量資料庫、MCP 與 A2A

> AIMock 在單一連接埠上模擬整個 AI 技術堆疊（LLM、向量資料庫、MCP、A2A），讓智能體測試套件在三秒內跑完，而且每次輸出都相同。

- Source: https://oharu121.com/zh-tw/blog/aimock-deterministic-testing-ai-agent-llm-vectordb-a2a/
- Published: 2026-08-10T18:10:57+09:00
- Tags: 測試, 生成式 AI, TypeScript, OpenAI, Python, Google Cloud

---
## 引言

測試建立在 LLM 之上的應用程式有個麻煩的問題：同一個提示送兩次，回來的內容會稍有不同。週一通過的測試，週二就失敗了。原因不是程式碼有 bug，而是模型的非決定性。

2026 年的智能體應用讓這件事更嚴重。單一請求會碰到六、七個服務：除了 LLM，還有做檢索增強生成（RAG）的向量資料庫、Model Context Protocol（MCP）工具，以及走 Agent2Agent（A2A）協定的智能體之間的流量。要怎麼模擬其中每一個，才是真正的問題。

AIMock 是 CopilotKit 在 2026 年 4 月釋出的開源模擬伺服器，它在單一連接埠上回答了以上全部。本文建立的測試套件，對著模擬的 LLM、向量資料庫與 A2A 服務跑 18 個案例只花 2.35 秒，每次的輸出都完全相同。

本文會走過 demo 智能體與它的測試、錄製與重播真實的測試固件、LLMock 的混沌注入 API，然後是另一半練習：把 Google codelab 的漢堡智能體從 a2a-sdk v0.2 遷移到 v1.0、部署到 Cloud Run，再讓同一個 TypeScript 用戶端指向它。

## AIMock 是什麼

[AIMock](https://github.com/CopilotKit/aimock) 是一個 npm 套件，用來模擬 AI 應用程式會對話的每一個服務。

- 11 種以上的 LLM 供應商（OpenAI、Claude、Gemini、Bedrock、Azure、Vertex AI、Ollama、Cohere 等）
- 完整支援 MCP JSON-RPC 2.0
- A2A 的智能體卡片探索與 server-sent events（SSE）串流
- 向量資料庫模擬（Pinecone、Qdrant、ChromaDB）
- 混沌測試：注入 500、格式錯誤的 JSON、串流中途斷線，既可以用自行架設的伺服器，也可以透過 `setChaos()` API
- 零相依套件，只用 Node.js 內建模組

概念是一個設定檔、一個連接埠，涵蓋整個 AI 技術堆疊。

### 它在同類工具中的位置

我去找了競品，發現這個利基幾乎沒有直接對手。

| 工具 | 定位 | 與 AIMock 的重疊 |
| --- | --- | --- |
| Keploy | 錄製真實流量並產生測試，通用 API | 不支援 AI 專屬協定（MCP、A2A、SSE 串流） |
| VCR.py / Polly.js | 傳統的 HTTP 錄製與重播 | 可以用來模擬 LLM，但沒有工具呼叫這類 AI 專屬功能 |
| Portkey / Helicone | 具備快取與重播的 LLM 閘道 | 面向正式環境，沒有故障注入，也不模擬 MCP |

如果要測試包含 MCP 與 A2A 的智能體應用程式，AIMock 目前是唯一選項。如果只需要模擬 OpenAI API，VCR 式的 HTTP 重播就夠了。

## 錄製與重播測試固件，以及它為什麼重要

錄製與重播是 AIMock 的核心功能之一：先擷取一次真實的 API 回應，之後在測試中原封不動地回放。

第一次看到它時，我想不通這跟 n8n 那種重跑工作流程中間步驟的做法差在哪裡。結果兩者要解的問題根本不同。

錄製與重播解決的是這個：

```text
# Monday: test passes
assert response.includes("SQL injection risk")  ✅

# Tuesday: same code, same prompt, OpenAI rephrased
assert response.includes("SQL injection risk")  ❌
# it returned "potential SQL vulnerability"
```

有了測試固件，一次 LLM 呼叫的行為就像純函式。相同輸入，每次都是相同輸出。

| 面向 | 沒有測試固件 | 有測試固件 |
| --- | --- | --- |
| 成本 | 每跑一次測試就打一次真實 API | 零 |
| 速度 | 每次 LLM 呼叫 2–30 秒 | 1 毫秒以內 |
| 決定性 | 每次輸出都不同 | 位元組層級完全相同 |
| CI 穩定度 | 會因網路、速率限制、API 變更而不穩 | 穩定 |
| 邊界情況 | 無法重現 | 永久保留下來 |

### 在 CI 用模擬是不是在逃避？

「用模擬跑的 CI 抓得到正式環境的問題嗎」是個合理的疑問。答案是：它測的本來就是別的東西。

| 你要測的是什麼 | 需要真實 API 嗎 |
| --- | --- |
| JSON 解析器處理得了這種回應結構嗎 | 不用，測試固件就夠 |
| 遇到 500 時重試邏輯有效嗎 | 不用，混沌注入更合適 |
| 這個提示能不能產出好的輸出 | **需要**，對著模擬做不到 |
| UI 有正確算繪出回應嗎 | 不用，測試固件就夠 |

實務上最好的模式是**兩層 CI 流水線**：

- **每個 PR**：跑得快的模擬測試，驗證程式邏輯
- **每晚或每週**：對真實 API 跑整合測試，做偏移偵測與提示驗證

AIMock 的**偏移偵測**正好補上這一塊。每天在 CI 跑一次，它會三方比對 SDK 的型別定義、真實 API 的回應，以及 AIMock 自己的輸出，在使用者發現之前就先抓到供應商改了 API。

## 打造 demo：智慧採購助理

為了把主要功能都用過一輪，我為一個採購智能體寫了測試。

### 架構

*Figure — Architecture: 一次請求會讓智能體向外呼叫四次，而每一次都落在 localhost 上的模擬服務。*

測試裡用到三樣東西：

- **LLMock**：模擬 OpenAI API（`@copilotkit/aimock` 的 `LLMock` 類別）
- **VectorMock**：相容 Pinecone 的向量資料庫模擬（`VectorMock` 類別）
- **A2A 模擬**：我用 `http.createServer` 自行實作的賣家智能體

### 專案結構

```text
aimock-a2a-demo/
├── src/
│   ├── buyer-agent/
│   │   ├── index.ts          # the main BuyerAgent orchestrator
│   │   ├── a2a-client.ts     # A2A protocol client
│   │   ├── knowledge-base.ts # vector DB / RAG layer
│   │   └── reasoner.ts       # LLM reasoning
│   └── types/
│       ├── a2a.ts            # A2A protocol types
│       └── product.ts        # product types
├── tests/
│   ├── setup.ts              # AIMock setup
│   ├── buyer-agent.test.ts   # main test suite
│   └── chaos.test.ts         # chaos tests
├── fixtures/llm/             # recorded LLM responses
├── aimock.json               # AIMock config
└── package.json
```

### Step 1：設定 LLMock

LLMock 以 `systemMessage` 或 `userMessage` 的模式比對來註冊測試固件。

```ts title="tests/setup.ts"
import { LLMock, VectorMock } from "@copilotkit/aimock";

const llmock = new LLMock({ port: 0 }); // random port

// route fixtures by the content of the system prompt
llmock.on({ systemMessage: "evaluating product offers" }, {
  content: JSON.stringify({
    topPickIndex: 0,
    reasoning: "The MacBook Air M4 offers the best combination...",
    confidence: 0.92,
  }),
});

llmock.on({ systemMessage: "purchasing concierge" }, {
  content: JSON.stringify({
    refinedQuery: "ultrabook laptop under 1.3kg...",
    priorities: ["weight under 1.3kg", "battery life 15+ hours"],
    priceRange: { min: 900, max: 1300 },
  }),
});

await llmock.start();
process.env.OPENAI_BASE_URL = `${llmock.url}/v1`;
process.env.OPENAI_API_KEY = "mock-key";
```

**這裡學到的事**：`onMessage()` 只比對使用者訊息的文字。當應用程式呼叫 LLM 超過一次時（這裡是兩次，一次規劃搜尋策略、一次評估報價），相似的使用者訊息可能會比對到非預期的測試固件。用 `on({ systemMessage: "..." })` 依系統提示的內容來分流才可靠。

**註冊順序同樣重要。** LLMock 會回傳第一個比對成功的測試固件，所以要先註冊比較明確的模式，把萬用的那個（`onMessage(/.*/)`）放最後。

### Step 2：設定 VectorMock

VectorMock 提供相容 Pinecone 的 API。

```ts title="tests/setup.ts"
const vectorMock = new VectorMock({ port: 0 });

vectorMock.addCollection("product-knowledge", { dimension: 1536 });

// register static results — the same results come back every time
vectorMock.onQuery("product-knowledge", [
  {
    id: "review-macbook-air",
    score: 0.94,
    metadata: {
      brand: "Apple",
      model: "MacBook Air M4",
      text: "MacBook Air M4: 1.24kg, 18hr battery life...",
    },
  },
  // ... more entries
]);

await vectorMock.start();
```

**這裡學到的事**：`QueryResult` 型別沒有 `text` 欄位。文字資料要放在 `metadata.text`，應用程式再以 `m.metadata?.text` 讀回來。

VectorMock 理論上可以透過 `mount()` 掛在跟 LLMock 同一個連接埠上，但路徑前綴不會被去掉，所以實際可行的做法是讓它跑在自己的連接埠。

### Step 3：模擬 A2A 協定

A2A 是 Google 與 Linux Foundation 推出的開放智能體間協定（v1.2，150 個以上的參與組織）。

它真正站得住腳的地方是**跨組織的智能體通訊**，也就是你的智能體去跟另一家公司的智能體對話，例如金流服務或物流智能體。在同一個應用程式內部的子智能體之間使用 A2A 是殺雞用牛刀，一次函式呼叫就夠了。

在這個 demo 裡，A2A 位於買家智能體與賣家智能體之間。它們設定上分屬不同組織，所以是正當的用法。

我用 `http.createServer` 自行實作了 A2A 模擬，遵循 A2A v1.0：

```ts title="tests/setup.ts"
const server = createServer((req, res) => {
  // Agent card discovery: GET /.well-known/agent-card.json (v1.0)
  if (req.url === "/.well-known/agent-card.json" && req.method === "GET") {
    res.writeHead(200, { "Content-Type": "application/json" });
    res.end(JSON.stringify({
      name: "Mock Electronics Seller",
      skills: [{ id: "product-search", name: "Product Search", ... }],
      ...
    }));
    return;
  }

  // JSON-RPC: POST / — SendMessage (v1.0)
  if (req.method === "POST") {
    const rpc = JSON.parse(body);
    if (rpc.method === "SendMessage") {
      res.end(JSON.stringify({
        jsonrpc: "2.0", id: rpc.id,
        result: {
          message: {
            role: "ROLE_AGENT",
            parts: [{ data: { offers: mockOffers } }],
            messageId: `reply-${rpc.id}`,
            contextId: rpc.params?.message?.contextId,
          },
        },
      }));
    }
  }
});
```

### Step 4：執行測試

18 個測試案例，全部在三秒內跑完。

```text
$ pnpm test

 ✓ tests/buyer-agent.test.ts (8 tests) 46ms
   ✓ A2A: Seller Agent Discovery > discovers seller agent card
   ✓ A2A: Seller Agent Discovery > checks skill availability
   ✓ A2A: Seller Agent Discovery > sends message and receives offers
   ✓ Vector DB: Knowledge Base > returns deterministic results
   ✓ Vector DB: Knowledge Base > same query twice returns identical results
   ✓ Vector DB: Knowledge Base > queryWithFallback returns empty on unreachable server
   ✓ Integration: Full Purchasing Flow > end-to-end: query → RAG → LLM → A2A → LLM
   ✓ Integration: Full Purchasing Flow > works without RAG knowledge

 ✓ tests/chaos.test.ts (10 tests) 2026ms
   ✓ Chaos: Vector DB failures > survives connection refused
   ✓ Chaos: Vector DB failures > respects timeout
   ✓ Chaos: Vector DB failures > throws on server error
   ✓ Chaos: LLM failures > malformed JSON
   ✓ Chaos: LLM failures > empty response
   ✓ Chaos: LLMock setChaos() > malformedRate — every response is garbled
   ✓ Chaos: LLMock setChaos() > dropRate — every request gets 500
   ✓ Chaos: LLMock setChaos() > clearChaos() restores normal operation
   ✓ Chaos: LLMock nextRequestError() > single request fails, next succeeds
   ✓ Resilience > agent degrades gracefully

 Test Files  2 passed (2)
      Tests  18 passed (18)
   Duration  2.35s
```

其中兩個值得拿出來看。

**證明決定性：**

```ts title="tests/buyer-agent.test.ts"
test("same query twice returns identical results (determinism proof)", async () => {
  const results1 = await kb.query("laptop");
  const results2 = await kb.query("laptop");

  // real Pinecone gives no such guarantee
  expect(results1).toEqual(results2);
});
```

**優雅降級：**

```ts title="tests/buyer-agent.test.ts"
test("works without RAG knowledge (graceful degradation)", async () => {
  const agent = new BuyerAgent({
    sellerAgentUrl: sellerUrl,
    knowledgeBase: {
      apiKey: "mock-key",
      baseUrl: "http://localhost:1", // unreachable → falls back
      indexName: "product-knowledge",
    },
    ragTimeoutMs: 500,
  });

  const result = await agent.findProduct({
    query: "I need a lightweight laptop for travel",
  });

  // even with RAG down, the agent still works via LLM + A2A
  expect(result.topPick).toBeDefined();
  expect(result.knowledgeUsed).toHaveLength(0);
});
```

這個測試沒有模擬基礎設施是很難寫的。總不能為了跑它就去把一個真實的 Pinecone 執行個體停掉。

## A2A 生態系的現狀

研究這個協定時，我去找了可以自由互動的公開智能體。

### 找得到的東西

- **Hello World Agent**：`https://hello-world-gxfr.onrender.com/.well-known/agent.json`。第一個公開的 A2A 智能體。只會回聲，但用來確認用戶端能不能動很有用
- **A2A Registry Playground**：[a2a-registry.org/playground](https://www.a2a-registry.org/playground)，可以跟已登錄的智能體對話的實驗場
- **Google codelab**：[Purchasing Concierge](https://codelabs.developers.google.com/intro-a2a-purchasing-concierge)，把兩個智能體部署到 Cloud Run 並讓它們走 A2A 對話的實作教學。在免費額度內就能跑完

### 率直的評估

| 已經存在的 | 還不存在的 |
| --- | --- |
| Hello world / 回聲類智能體 | 真實企業公開的正式 A2A 端點 |
| Google codelab 的示範 | Stripe、Twilio 這類廠商的 A2A API |
| 社群的玩具型智能體 | 具備真正實用能力的智能體市集 |

A2A 有 150 個以上的參與組織、GitHub 星數超過 22,000，但**幾乎沒有可以自由互動又真正有用的公開智能體**。企業端的 A2A 採用（Azure AI Foundry、Bedrock AgentCore）幾乎全部躲在認證後面。

協定本身已經成熟（v1.2，隸屬 Linux Foundation）。智能體還沒跟上。

## 在本機跑起 Google codelab 的真實 A2A 智能體

前半部用的是以 `http.createServer` 寫成的 A2A 賣家智能體模擬。下一步是在本機跑一個真實的 A2A 智能體：Google [Purchasing Concierge](https://codelabs.developers.google.com/intro-a2a-purchasing-concierge) codelab 裡的那一個。

### 複製儲存庫並調整

```bash
git clone https://github.com/alphinside/purchasing-concierge-intro-a2a-codelab-starter.git
cd purchasing-concierge-intro-a2a-codelab-starter/remote_seller_agents/burger_agent
```

這個漢堡智能體是建立在 CrewAI、LiteLLM 與 Vertex AI Gemini 之上的 A2A 賣家智能體。它接收使用者的漢堡訂單並處理。

相對於原始儲存庫，需要四處調整。

**1. 改用 Python 3.13**

```toml title="pyproject.toml"
requires-python = ">=3.13"
```

```dockerfile title="Dockerfile"
FROM python:3.13-slim
```

**2. 重新產生 `uv.lock`**

`uv.lock` 會把每個套件的完整下載 URL 寫死，因此它帶著產生它的那台機器所指向的索引來源。我的那份是在安全代理鏡像後面產生的，鎖定檔裡每一個 URL 都繞過那個鏡像。在連不到該鏡像的環境下，下載就會逾時。

```bash
rm uv.lock
UV_DEFAULT_INDEX=https://pypi.org/simple/ uv lock --python 3.13
```

這是因為 `uv sync --frozen` [會直接沿用鎖定檔裡寫死的 URL](https://github.com/astral-sh/uv/issues/19625)。改索引設定並不會改掉鎖定檔裡既有的 URL，只能重新產生。

在專案內也一併覆寫索引是值得的：

```toml title="pyproject.toml"
[[tool.uv.index]]
url = "https://pypi.org/simple/"
default = true
```

**3. 把 `litellm` 加成明確的相依套件**

`agent.py` 裡有 `import litellm`，但 `litellm` 並不在 `pyproject.toml` 的相依清單中。它以前是靠 CrewAI 的遞移相依帶進來的，新版本已經拿掉了。

```bash
UV_DEFAULT_INDEX=https://pypi.org/simple/ uv add litellm
```

### 把 a2a-sdk 從 v0.2 遷移到 v1.0

原始程式碼寫的是 a2a-sdk v0.2，但 `>=0.2.16` 這個限制會安裝到 v1.1.2，它的破壞性變更會產生 import 錯誤。我把程式碼更新到 v1.0。

主要變更：

| 項目 | v0.2 | v1.0 |
| --- | --- | --- |
| 伺服器建構 | `A2AStarletteApplication` 類別 | `create_jsonrpc_routes()` / `create_agent_card_routes()` 路由工廠 |
| AgentCard | `url` 欄位 | `supported_interfaces` 清單（`AgentInterface`） |
| RequestHandler | 沒有 `agent_card` | `agent_card` 為必填 |
| Part 建構 | `Part(root=TextPart(text="..."))` | `Part(text="...")`（以 Protobuf 為基礎） |
| JSON-RPC 方法 | `tasks/send` | `SendMessage` |
| 智能體卡片路徑 | `/.well-known/agent.json` | `/.well-known/agent-card.json` |
| 協定版本 | 無（隱含 0.3） | 必須帶 `A2A-Version: 1.0` 標頭 |
| 錯誤型別 | `ServerError` | 直接使用 `UnsupportedOperationError` |

**`__main__.py`，伺服器建構：**

```python title="__main__.py"
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.types import AgentCard, AgentInterface, AgentCapabilities, AgentSkill
from starlette.applications import Starlette

agent_card = AgentCard(
    name="burger_seller_agent",
    description="Helps with creating burger orders",
    version="1.0.0",
    capabilities=AgentCapabilities(streaming=True),
    supported_interfaces=[
        AgentInterface(protocol_binding="JSONRPC", url=agent_host_url)
    ],
    skills=[skill],
)

request_handler = DefaultRequestHandler(
    agent_executor=BurgerSellerAgentExecutor(),
    task_store=InMemoryTaskStore(),
    agent_card=agent_card,  # required in v1.0
)

routes = []
routes.extend(create_agent_card_routes(agent_card))
routes.extend(create_jsonrpc_routes(request_handler, "/"))

app = Starlette(routes=routes)
uvicorn.run(app, host=host, port=port)
```

**`agent_executor.py`，以 Protobuf 為基礎的型別：**

```python title="agent_executor.py"
from a2a.types import Message, Part, Role, Task
from a2a.utils.errors import UnsupportedOperationError

class BurgerSellerAgentExecutor(AgentExecutor):
    async def execute(self, context, event_queue):
        query = context.message.parts[0].text if context.message else ""
        result = await self.agent.invoke(query, context.context_id)

        # v1.0: construct directly with Part(text="..."), no TextPart wrapper
        await event_queue.enqueue_event(
            Message(
                role=Role.ROLE_AGENT,
                parts=[Part(text=str(result))],
                message_id=context.task_id or "msg",
                context_id=context.context_id,
            )
        )
```

**`agent.py`，改成非同步：**

CrewAI 的 `kickoff()` 是同步的，但在 a2a-sdk v1.0 裡，請求處理器是從非同步情境中被呼叫的。在非同步情境內呼叫同步方法會拋出 `Agent execution was invoked synchronously from within a running event loop`。

```python title="agent.py"
# Before: def invoke(self, query, sessionId) -> str:
async def invoke(self, query, sessionId) -> str:
    # ...
    # Before: response = crew.kickoff(inputs)
    response = await crew.kickoff_async(inputs)
    return response
```

### 確認可以動

```bash
# the Vertex AI API has to be enabled
gcloud services enable aiplatform.googleapis.com

# start it locally
uv run . --host 127.0.0.1 --port 8080
```

確認智能體卡片：

```bash
$ curl http://127.0.0.1:8080/.well-known/agent-card.json | jq .name
"burger_seller_agent"
```

用 A2A v1.0 送一則訊息：

```bash
$ curl -X POST http://127.0.0.1:8080/ \
  -H "Content-Type: application/json" \
  -H "A2A-Version: 1.0" \
  -d '{"jsonrpc":"2.0","id":"1","method":"SendMessage",
       "params":{"message":{"role":"ROLE_USER",
       "parts":[{"text":"What burgers do you have?"}],
       "messageId":"m1","contextId":"ctx1"}}}'
```

回應：

```json
{
  "result": {
    "message": {
      "role": "ROLE_AGENT",
      "parts": [{
        "text": "We have Classic Cheeseburger for IDR 85K, Double Cheeseburger for IDR 110K, Spicy Chicken Burger for IDR 80K, and Spicy Cajun Burger for IDR 85K. Do any of these sound good?"
      }]
    }
  },
  "id": "1",
  "jsonrpc": "2.0"
}
```

整條路徑都通了：Vertex AI Gemini → CrewAI → A2A v1.0。

### 部署到 Cloud Run

本機跑起來之後，下一步是 Cloud Run。這個漢堡智能體是由 uvicorn 啟動的 Starlette ASGI 伺服器，因此合適的是 **Cloud Run** 而不是 Cloud Functions。Cloud Functions 每個請求呼叫一個 Python 函式；A2A 智能體則是一個常駐的 HTTP 伺服器，帶著數個 Starlette 路由（`/.well-known/agent-card.json` 與位於 `/` 的 JSON-RPC 端點）。

```bash
gcloud run deploy burger-agent \
  --source . \
  --port 8080 \
  --allow-unauthenticated \
  --region us-central1 \
  --min-instances 0 \
  --max-instances 1 \
  --memory 1Gi
```

`--source .` 會在 Cloud Build 上執行 Dockerfile 建置，把映像檔存到 Artifact Registry，再部署到 Cloud Run。

部署後要設一個環境變數，讓智能體卡片的 `url` 欄位反映 Cloud Run 的 URL：

```bash
gcloud run services update burger-agent \
  --region us-central1 \
  --set-env-vars HOST_OVERRIDE=https://burger-agent-XXXXXXXXXX.us-central1.run.app
```

確認結果：

```bash
$ curl https://burger-agent-XXXXXXXXXX.us-central1.run.app/.well-known/agent-card.json | jq .name
"burger_seller_agent"

$ curl -X POST https://burger-agent-XXXXXXXXXX.us-central1.run.app/ \
  -H "Content-Type: application/json" \
  -H "A2A-Version: 1.0" \
  -d '{"jsonrpc":"2.0","id":"1","method":"SendMessage",
       "params":{"message":{"role":"ROLE_USER",
       "parts":[{"text":"What burgers do you have?"}],
       "messageId":"m1","contextId":"ctx1"}}}'
```

```json
{
  "result": {
    "message": {
      "role": "ROLE_AGENT",
      "parts": [{
        "text": "We have Classic Cheeseburger for IDR 85K, Double Cheeseburger for IDR 110K, Spicy Chicken Burger for IDR 80K, and Spicy Cajun Burger for IDR 85K. What would you like to order?"
      }]
    }
  }
}
```

**關於成本**：設定 `--min-instances 0` 之後，執行個體在請求之間會縮到零，不會計費。代價是 10–15 秒的冷啟動，對 demo 或測試來說沒問題。

## 讓 TypeScript 買家智能體支援 A2A v1.0

要連上 Cloud Run 上的真實賣家智能體，TypeScript 的 A2A 用戶端也得做 v1.0 的調整。

### A2A 用戶端的變更

```ts title="src/buyer-agent/a2a-client.ts"
export class A2AClient {
  // v1.0: agent-card.json (was: agent.json)
  async discover(): Promise<AgentCard> {
    const url = `${this.agentUrl}/.well-known/agent-card.json`;
    const res = await fetch(url);
    this.card = await res.json();
    return this.card;
  }

  // v1.0: SendMessage plus the A2A-Version header
  async sendMessage(text: string, contextId?: string): Promise<A2AMessage> {
    const rpcRequest = {
      jsonrpc: "2.0",
      id: messageId,
      method: "SendMessage",  // was: "tasks/send"
      params: {
        message: {
          role: "ROLE_USER",  // was: "user"
          parts: [{ text }],  // was: { type: "text", text }
          messageId,
          contextId,
        },
      },
    };

    const res = await fetch(this.agentUrl, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        "A2A-Version": "1.0",  // required in v1.0
      },
      body: JSON.stringify(rpcRequest),
    });

    const rpcResponse = await res.json();
    return rpcResponse.result.message;  // was: result (a task object)
  }
}
```

### v0.3 到 v1.0 的變更整理

| 項目 | v0.3 | v1.0 |
| --- | --- | --- |
| 智能體卡片路徑 | `/.well-known/agent.json` | `/.well-known/agent-card.json` |
| JSON-RPC 方法 | `tasks/send` | `SendMessage` |
| role 的值 | `"user"` / `"agent"` | `"ROLE_USER"` / `"ROLE_AGENT"` |
| Part 結構 | `{ type: "text", text: "..." }` | `{ text: "..." }` |
| 標頭 | 無 | `A2A-Version: 1.0` |
| 回應 | `result` 是一個 Task（訊息陣列） | `result.message` 是單一 Message |

### 連上 Cloud Run 上的真實智能體

我確認了 TypeScript 的 A2A 用戶端可以跟 Cloud Run 上的真實漢堡智能體對話：

```ts
const client = new A2AClient("https://burger-agent-XXXXXXXXXX.us-central1.run.app");

const card = await client.discover();
// → Name: burger_seller_agent, Skills: ["create_burger_order"]

const reply = await client.sendMessage("What burgers do you have?");
// → role: ROLE_AGENT
// → "We have Classic Cheeseburger for IDR 85K, Double Cheeseburger for IDR 110K, ..."
```

模擬的測試與真實的智能體，走的是同一份 A2A 用戶端程式碼與同一個 v1.0 協定。測試指向 AIMock 的模擬伺服器，正式環境指向真實的 Cloud Run 智能體。改的只有 URL。

## 實際錄製測試固件

抽象地談完錄製與重播之後，我實際跑了一次。

### 啟動 llmock

`@copilotkit/aimock` 套件附帶一個名為 `llmock` 的 CLI 執行檔。`--record` 旗標會讓它以錄製代理伺服器的形式啟動。

```bash
# start llmock in record mode, listening on port 4010
pnpm mock:record
# → llmock --record --provider-openai https://api.openai.com -f ./fixtures/llm
```

### 讓應用程式經由 llmock 執行

把 `OPENAI_BASE_URL` 指向 llmock，應用程式中每一次 OpenAI SDK 呼叫就都會經過它。應用程式從 `.env` 讀取 API 金鑰，llmock 則透明地轉發給 OpenAI。

```bash
OPENAI_BASE_URL=http://localhost:4010/v1 \
SELLER_URL=https://burger-agent-XXXXXXXXXX.us-central1.run.app \
pnpm dev
```

### 錄下來的測試固件

像這樣的 JSON 會出現在 `fixtures/llm/recorded/`：

```json
{
  "fixtures": [{
    "match": {
      "userMessage": "User request: \"I need a lightweight laptop...\"",
      "model": "gpt-4o-mini",
      "turnIndex": 0,
      "hasToolResult": false
    },
    "response": {
      "content": "{\"refinedQuery\": \"lightweight travel laptop under $1200\", ...}"
    },
    "metadata": {
      "systemHash": "a11524e7"
    }
  }]
}
```

`match` 底下的欄位（`userMessage`、`model`、`turnIndex`）就是比對條件。從此之後，只要請求符合同樣的條件，就會拿到這份回應，不會再打到 OpenAI。

### 實際用起來的感覺

流程之簡單是最令人意外的地方：

1. `pnpm mock:record` 啟動代理伺服器
2. `OPENAI_BASE_URL=http://localhost:4010/v1 pnpm dev` 照常執行應用程式
3. 測試固件自己存進 `fixtures/llm/recorded/`
4. 之後 `pnpm mock:start` 就會重播它們

**完全不用改程式碼。** 改的只有 `OPENAI_BASE_URL` 指向哪裡。

## LLMock 的混沌注入

上面那次執行裡有四個測試（`setChaos()` 與 `nextRequestError()` 那幾個）到目前為止還沒解釋。它們用的是 LLMock 的**程式化混沌注入 API**，可以重現 LLM 的失敗情境，不必像前五個混沌測試那樣自己架一台 HTTP 伺服器來偽造 500 或逾時。

### setChaos()：機率性的故障注入

`setChaos()` 設定的是套用到每個請求的失敗機率。

```ts title="tests/chaos.test.ts"
test("setChaos({ malformedRate: 1 }) — every response is garbled", async () => {
  const mock = new LLMock({ port: 0 });

  // register a valid fixture (/.*/ matches every message)
  mock.onMessage(/.*/, {
    content: '{"refinedQuery": "test", "priorities": [], "priceRange": {"min": 0, "max": 100}}',
  });

  // chaos: corrupt the response 100% of the time
  mock.setChaos({ malformedRate: 1.0 });

  await mock.start();
  process.env.OPENAI_BASE_URL = `http://localhost:${mock.port}/v1`;

  const reasoner = new Reasoner("gpt-4o-mini");

  // the fixture matches, but chaos breaks the response
  await expect(
    reasoner.buildSearchStrategy("laptop", [])
  ).rejects.toThrow();

  await mock.stop();
});
```

`ChaosConfig` 有三個機率參數：

| 參數 | 效果 | 沒有測試固件也能用 |
| --- | --- | --- |
| `dropRate` | 回傳 500 | 可以 |
| `malformedRate` | 回傳 `{malformed json: <<<chaos>>>}` | 不行，需要比對成功 |
| `disconnectRate` | 在串流中途切斷 TCP 連線 | 不行，需要比對成功 |

有個陷阱：`onMessage("*")` 比對的是**字面上的字串** `*`，不是萬用字元。要比對所有訊息得用正規表示式 `/.*/`。

### clearChaos()：把它關掉

`clearChaos()` 會移除混沌設定，恢復正常的測試固件回應。適合用來測試從失敗中恢復。

```ts title="tests/chaos.test.ts"
test("clearChaos() restores normal operation", async () => {
  const mock = new LLMock({ port: 0 });
  mock.onMessage(/.*/, { content: validJson });
  mock.setChaos({ malformedRate: 1.0 });

  await mock.start();
  // ... env setup ...

  const reasoner = new Reasoner("gpt-4o-mini");

  // chaos on — it fails
  await expect(
    reasoner.buildSearchStrategy("laptop", [])
  ).rejects.toThrow();

  // chaos off — it works
  mock.clearChaos();

  const strategy = await reasoner.buildSearchStrategy("laptop", []);
  expect(strategy.refinedQuery).toBe("laptop"); // ✓ valid JSON
});
```

### nextRequestError()：一次性的錯誤注入

`nextRequestError()` 只會讓**下一個請求**回傳錯誤，然後就自我消耗。很適合「失敗一次、重試就成功」的情境。

```ts title="tests/chaos.test.ts"
test("nextRequestError(400) — single request fails, next succeeds", async () => {
  const mock = new LLMock({ port: 0 });
  mock.onMessage(/.*/, { content: validJson });

  // one-shot: return 400 for the next request only
  mock.nextRequestError(400, { message: "Bad Request" });

  await mock.start();
  // ... env setup ...

  const reasoner = new Reasoner("gpt-4o-mini");

  // first call: 400
  await expect(
    reasoner.buildSearchStrategy("laptop", [])
  ).rejects.toThrow();

  // second call: the one-shot is spent → normal response
  const strategy = await reasoner.buildSearchStrategy("laptop", []);
  expect(strategy.refinedQuery).toBe("laptop"); // ✓
});
```

要注意 OpenAI SDK 會自動重試 5xx 錯誤（500、503 等）。用一次性注入可靠地測試錯誤時，要用不會被重試的 4xx。

### 測試執行結果

```text
$ pnpm test

 ✓ tests/buyer-agent.test.ts (8 tests) 42ms
 ✓ tests/chaos.test.ts (10 tests) 1918ms
   ✓ Chaos: Vector DB failures > survives connection refused
   ✓ Chaos: Vector DB failures > respects timeout
   ✓ Chaos: Vector DB failures > throws on server error
   ✓ Chaos: LLM failures > malformed JSON
   ✓ Chaos: LLM failures > empty response
   ✓ Chaos: LLMock setChaos() > malformedRate — every response is garbled
   ✓ Chaos: LLMock setChaos() > dropRate — every request gets 500
   ✓ Chaos: LLMock setChaos() > clearChaos() restores normal operation
   ✓ Chaos: LLMock nextRequestError() > single request fails, next succeeds
   ✓ Resilience > agent degrades gracefully

 Test Files  2 passed (2)
      Tests  18 passed (18)
   Duration  2.2s
```

兩種做法都可行：自行架設 HTTP 伺服器的混沌測試（前五個），以及 LLMock 內建的混沌 API（接下來四個；最後一個測試 `agent degrades gracefully` 兩邊都不屬於）。API 版本的程式碼比較少，意圖也更清楚。

## 總結

我用 AIMock 建立了一套 AI 智能體測試，接著把 Google codelab 的真實 A2A 智能體遷移到 a2a-sdk v1.0、在本機跑起來、部署到 Cloud Run，再從 TypeScript 買家智能體連上它。

**AIMock 站得住腳的地方：**

- 測試同時結合 LLM、向量資料庫與其他數個服務的智能體應用程式
- 在 CI/CD 中快速、便宜、穩定地跑測試
- 透過故障注入驗證容錯能力

**殺雞用牛刀的地方：**

- 只要模擬 OpenAI API 的情況，VCR 式的 HTTP 重播就夠了
- 評估提示的品質，這件事對著模擬做不到

**AIMock 的實作筆記：**

- `onMessage()` 只比對使用者訊息。`on({ systemMessage })` 才能依系統提示分流
- `onMessage("*")` 比對的是字面上的 `*`，不是萬用字元。要比對全部請用 `onMessage(/.*/)`
- 測試固件的註冊順序有影響：先比對到的先勝出
- VectorMock 的 `QueryResult` 沒有 `text` 欄位，文字要放進 `metadata`
- VectorMock 要跑在跟 LLMock 分開的自有連接埠上
- `setChaos()` 的 `malformedRate` 與 `disconnectRate` 需要測試固件比對成功，只有 `dropRate` 不需要
- OpenAI SDK 會自動重試 5xx。用 `nextRequestError()` 測試錯誤時要用 4xx

**a2a-sdk v1.0 遷移與部署的心得：**

- `A2AStarletteApplication` 已經消失。改用路由工廠（`create_jsonrpc_routes` / `create_agent_card_routes`）搭配標準的 Starlette 應用
- 型別系統從 Pydantic 換成 Protobuf。直接建構即可，例如 `Part(text="...")`
- 智能體卡片路徑從 `agent.json` 改成 `agent-card.json`
- JSON-RPC 方法從 `tasks/send` 改成 `SendMessage`，而且必須帶 `A2A-Version: 1.0` 標頭
- 寫死在 `uv.lock` 裡的 PyPI 鏡像 URL，用 `UV_DEFAULT_INDEX` 重新產生就能解決
- A2A 智能體是 Starlette ASGI 伺服器，所以要部署到 Cloud Run（Docker 容器），而不是 Cloud Functions（單一函式處理模型）
- 在 Cloud Run 上，`HOST_OVERRIDE` 環境變數決定智能體卡片對外的 URL
- TypeScript 的 A2A 用戶端需要同樣的 v1.0 調整：端點路徑、方法名稱、role 的值、標頭
- 因為模擬與真實智能體共用同一份用戶端程式碼與協定，在測試與正式環境之間切換只是改一個 URL

## 參考連結

- [AIMock GitHub 儲存庫](https://github.com/CopilotKit/aimock)
- [AIMock 文件](https://aimock.copilotkit.dev/docs/)
- [A2A Protocol](https://a2a-protocol.org)
- [Google codelab：A2A Purchasing Concierge](https://codelabs.developers.google.com/intro-a2a-purchasing-concierge)
- [A2A Registry](https://www.a2a-registry.org/)