使用 AgentCore 与 Strands 构建：运行时主权风险代理人开发者工作坊

目标受众： 专业 Python/AWS 开发者

时长： 2 小时

主要 AWS AI 服务： Amazon Bedrock AgentCore Runtime、Strands Agents、Amazon Bedrock 模型

工作坊风格： 开发者实战动手构建，无需演示

项目输出： 一个具备本地验证、部署脚本、boto3 调用客户端、串流变体、会话清理以及大容量 Payload 处理器的全功能运行时代理人（Runtime Agent）。

本工作坊仅供教育工程教学使用。示例领域为主权风险分析，但输出内容不构成任何财务、投资、法律、税务或交易建议。

工作坊摘要

开发者将使用 Strands 与 AgentCore Runtime 构建一个生产级的主权风险代理人。工作坊内容涵盖决定性（Deterministic）计算工具、Payload 验证、冒烟测试、部署、boto3 客户端调用、串流响应、大容量 base64 Payload 处理以及会话清理。参与者在结束时将获得可重复使用的运行时模式，用于构建安全、具备可观测性且感知会话（Session-aware）的 AI 服务，并在 AWS 上将模型推理与可测试的 Python 逻辑相结合，以满足现代企业开发者的工作流程。

---

1. 开发者学习目标

在本工作坊结束时，开发者将能够：

● 构建一个由 Amazon Bedrock 模型支持的 Strands 代理人。

● 为以 LLM 驱动的代理人添加决定性的 Python 工具。

● 将代理人封装至 Amazon Bedrock AgentCore Runtime 进入点（Entrypoint）。

● 使用 AgentCore 入门工具包（Starter Toolkit）封装并启动运行时。

● 使用稳定的会话 ID（Session ID）通过 boto3 调用运行时。

● 为需要实时响应的客户端实施串流运行时进入点。

● 构建一个可处理 base64 编码的 Excel 和图片输入的大容量 Payload 进入点。

● 在部署前加入本地 Payload 契约验证与冒烟测试。

---

2. 本工作坊构建的架构

開發者筆記型電腦
  ├─ app.py                         # 同步 AgentCore Runtime 進入點
  ├─ app_streaming.py               # 串流 Runtime 進入點
  ├─ app_large_payload.py           # 多模態 / 大容量 Payload Runtime 進入點
  ├─ deploy_runtime.py              # 封裝並啟動執行期
  ├─ invoke_runtime.py              # boto3 用戶端呼叫
  ├─ stop_session.py                # 顯式工作階段清理
  ├─ validate_payload.py            # 本地 Payload 合約驗證
  ├─ smoke_test.py                  # 本地冒煙測試
  ├─ prompts/system.md              # 受控的系統提示詞
  └─ requirements.txt

AWS
  ├─ 透過 Strands 存取的 Amazon Bedrock 模型
  ├─ Amazon Bedrock AgentCore Runtime
  ├─ 由工具包建立的 Amazon ECR 映像檔
  ├─ IAM 執行角色
  └─ 雲端日誌 / 執行期輸出

---

3. 2 小时实战议程

---

4. 先决条件

python --version     # 建議版本：3.11+
aws --version
aws sts get-caller-identity
export AWS_DEFAULT_REGION=us-east-1

建立并启用虚拟环境：

python -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip

---

步骤 1 — 建立项目基础架构

开发者操作

mkdir -p agentcore-strands-runtime/prompts
cd agentcore-strands-runtime
cat > requirements.txt <<'EOF'
bedrock-agentcore
bedrock-agentcore-starter-toolkit
strands-agents
strands-agents-tools
boto3
botocore
pytest
EOF

建立系统提示词：

cat > prompts/system.md <<'EOF'
您是一位專業的主權風險工程助理。
請分析殖利率價差、信用利差擴大、主權債務重新定價、外匯壓力、
資金流動性、存續期間需求、資本流動以及防禦性輪動。
請務必包含以下章節：
1. 觀察 (Observation)
2. 推理 (Reasoning)
3. 風險隱含意義 (Risk implication)
4. 避險考量 (Hedge consideration)
5. 確認訊號 (Confirmation signals)
6. 失效觸發條件 (Invalidation triggers)
7. 限制 (Limitations)
請使用工具進行數值計算。請勿提供投資建議或自主交易指令。
EOF

安装依赖项目：

pip install -r requirements.txt

系统设计决策

● 提示词文件作为版本控制的行为： 系统提示词是运行时契约的一部分。将其保存在 prompts/system.md 中，可让开发者像审查代码变更一样审查行为变更。这非常重要，因为代理人的输出风格、安全边界和必要章节会直接影响下游客户端、测试与评估。

● 明确的依赖文件： 运行时部署必须具备可重复性。requirements.txt 定义了运行时所需的精确 Python 软件包界面。这也简化了部署封装、容器构建以及会话的疑难排解，因为每位参与者安装的依赖项目完全相同。

● 云部署前的微型架构： 开发者先在本地建立文件并验证行为，然后才部署到云。这减少了云调试的杂音。如果缺少提示词文件或 Python 导入失败，错误会在运行时被封装和启动之前在本地被提取到。

预期结果

项目中包含一个依赖文件和一个受控的系统提示词，该提示词将由 Strands 代理人读取。

---

步骤 2 — 构建同步 Strands 代理人运行时

开发者操作

建立 app.py。

from strands import Agent, tool
from strands.models import BedrockModel
from strands_tools import calculator
from bedrock_agentcore.runtime import BedrockAgentCoreApp

app = BedrockAgentCoreApp()

with open("prompts/system.md", "r", encoding="utf-8") as f:
    SYSTEM_PROMPT = f.read()

@tool
def yield_spread_bps(us_yield: float, peer_yield: float) -> str:
    """計算兩個殖利率之間的基點（bps）價差。"""
    spread = (us_yield - peer_yield) * 100
    return f"殖利率價差為 {spread:.1f} 基點。"

@tool
def fx_hedge_notional(exposure_usd: float, hedge_ratio: float) -> str:
    """根據 USD 風險敞口和避險比率計算避險名目本金。"""
    hedge = exposure_usd * hedge_ratio
    return f"外匯避險名目本金為 USD {hedge:,.2f}。"

@tool
def stress_loss_notional(position_notional: float, spread_move_bps: float, duration: float) -> str:
    """估算由價差變動引起的存續期間驅動價格影響。"""
    loss = position_notional * duration * (spread_move_bps / 10000)
    return f"預估存續期間影響為 USD {loss:,.2f}。"

agent = Agent(
    model=BedrockModel(
        model_id="amazon.nova-pro-v1:0",
        temperature=0.2,
        max_tokens=4000,
    ),
    tools=[calculator, yield_spread_bps, fx_hedge_notional, stress_loss_notional],
    system_prompt=SYSTEM_PROMPT,
)

@app.entrypoint
def sovereign_runtime(payload, context):
    prompt = payload.get("prompt", "")
    if not prompt.strip():
        return {"error": "缺少必要欄位：prompt"}

    request_id = payload.get("request_id", context.session_id)
    request = (
        f"請求 ID: {request_id}\n"
        f"執行期工作階段: {context.session_id}\n"
        f"使用者請求: {prompt}"
    )
    response = agent(request)
    return response.message["content"][0]["text"]

if __name__ == "__main__":
    app.run()

商业逻辑

代理人分析主权风险提示词，并使用工具计算基点价差、外汇避险名目本金以及存续期间压力影响。模型处理定性综合分析，而决定性计算则留在 Python 中执行。

代码逻辑

● BedrockAgentCoreApp() 建立运行时应用程序包装器。

● BedrockModel() 设置 Strands 所使用的 Amazon Bedrock 模型。

● @tool 将 Python 函数公开给 Strands 代理人。

● @app.entrypoint 标记可调用的运行时处理器（Handler）。

● 处理器验证 prompt，加入请求与会话上下文，调用代理人，并返回最终文字。

预期结果

执行 python app.py 会启动一个本地与 AgentCore 兼容的运行时处理程序。部署后，同一个进入点将接收来自 AgentCore Runtime 的 JSON Payload。

系统设计决策

● 用于定量逻辑的决定性工具： LLM 非常适合用于综合分析，但财务计算应该是决定性且可测试的。通过将殖利率价差、避险名目本金和存续期间压力数学运算移至 Python 工具中，开发者可以获得可重复的输出，并能独立于模型行为之外对计算进行单元测试。

● 请求 ID 与会话上下文： 进入点将请求与会话 ID 注入模型提示词中。这为响应提供了运作上下文，并有助于工程师关联运行时日志、客户端调用和用户工作流程。这也为生产环境的追踪（Tracing）建立了一个乾净的模式。

● 低温度的 Bedrock 模型： 代理人使用 temperature=0.2 来减少响应的变异性。专业的开发者工作坊应该教授稳定的工程模式，而非创意的提示词实验。较低的变异性有助于改善冒烟测试、展示以及未来的评估基准。

---

步骤 3 — 加入本地 Payload 验证

开发者操作

建立 validate_payload.py。

REQUIRED_FIELDS = ["prompt"]
OPTIONAL_FIELDS = ["request_id", "user_id", "excel_data", "image_data"]

def validate_payload(payload: dict) -> tuple[bool, list[str]]:
    errors = []
    for field in REQUIRED_FIELDS:
        if field not in payload or not str(payload[field]).strip():
            errors.append(f"缺少必要欄位：{field}")

    unknown = set(payload.keys()) - set(REQUIRED_FIELDS) - set(OPTIONAL_FIELDS)
    for field in sorted(unknown):
        errors.append(f"未知欄位：{field}")

    return len(errors) == 0, errors

if __name__ == "__main__":
    sample = {"prompt": "分析美中殖利率價差", "request_id": "demo-001"}
    ok, errors = validate_payload(sample)
    print("有效性", ok)
    print("錯誤", errors)

商业逻辑

验证器在运行时调用之前强制执行请求契约。这可以防止客户端传送格式错误的请求，并使调试更容易。

代码逻辑

该函数检查必要字段、拒绝未知字段，并返回一个布尔值与错误列表。它可以被测试、客户端或运行时代码导入。

预期结果

python validate_payload.py
# 有效性 True
# 錯誤 []

系统设计决策

● 部署前的 Payload 契约： 开发者应在调用云运行时之前在本地验证 API 契约。这减少了因缺少提示词或字段拼错而导致的运行时调用失败。这也为以后建立更强大的纲要（Schema）奠定了基础。

● 明确的未知字段检测： 拒绝未知字段有助于及早发现整合错误。如果没有这项机制，客户端可能会误以为元数据（Metadata）已被使用，而运行时其实默默忽略了它。清晰的验证能改善开发者反馈。

● 可重复使用的验证模块： 验证器是一个模块，而不仅仅是一个脚本。它可以在单元测试、CLI 客户端和运行时进入点中重复使用。这避免了在多个文件中重复编写契约逻辑。

---

步骤 4 — 加入冒烟测试

开发者操作

建立 smoke_test.py。

from validate_payload import validate_payload
from app import yield_spread_bps, fx_hedge_notional, stress_loss_notional

def test_payload_validation():
    ok, errors = validate_payload({"prompt": "分析價差"})
    assert ok
    assert errors == []

def test_missing_prompt_fails():
    ok, errors = validate_payload({"request_id": "x"})
    assert not ok
    assert "缺少必要欄位：prompt" in errors

def test_tool_outputs():
    assert "210.0" in yield_spread_bps(4.25, 2.15)
    assert "15,000,000.00" in fx_hedge_notional(25_000_000, 0.6)
    assert "125,000.00" in stress_loss_notional(10_000_000, 25, 5)

执行：

pytest -q

商业逻辑

冒烟测试在部署代理人之前，验证请求契约与决定性计算工具。

代码逻辑

测试直接导入验证与工具函数。它们不会调用模型，从而保持测试的高速与决定性。

预期结果

3 passed

系统设计决策

● 优先测试决定性部分： 模型的输出可能会有所不同，但工具数学运算和 Payload 验证不应该改变。测试决定性组件可以为开发者提供快速反馈，并在云部署前隔离错误。

● 冒烟测试中无模型依赖： 冒烟测试不会调用 Amazon Bedrock。这避免了成本、凭证、模型延迟和非决定性断言（Assertions）的问题。其目标是验证本地软件契约。

● 工作坊友善的失败消息： 简单的断言使失败原因一目了然。在两小时的工作坊中，调试必须快速且集中。这些测试能及早发现常见的设置与逻辑错误。

---

步骤 5 — 部署 AgentCore Runtime

开发者操作

建立 deploy_runtime.py。

from boto3.session import Session
from bedrock_agentcore_starter_toolkit import Runtime

region = Session().region_name or "us-east-1"

runtime = Runtime()
runtime.configure(
    entrypoint="app.py",
    auto_create_execution_role=True,
    auto_create_ecr=True,
    requirements_file="requirements.txt",
    region=region,
    agent_name="sovereign-runtime-hands-on",
)

result = runtime.launch()
print("AGENTCORE_RUNTIME_ARN=", result.agent_arn)
print("AGENTCORE_RUNTIME_ID=", result.agent_id)

执行：

python deploy_runtime.py
export AGENTCORE_RUNTIME_ARN="貼上代理人執行期 ARN"

商业逻辑

本地代理人成为一个受管的运行时端点，应用程序可以调用该端点。

代码逻辑

Runtime().configure() 封装进入点与依赖项目。launch() 部署运行时并返回标识符。

预期结果

脚本印出 AgentCore Runtime ARN 与运行时 ID。

系统设计决策

● 利用工具包提高实战速度： 入门工具包处理了部署机制，使工作坊能够专注于运行时设计，而非容器内部细节。开发者仍然可以看到产生的运行时 ARN，并可以在事后检查生成的 AWS 资源。

● 先部署单一运行时进入点： 仅部署 app.py 可以保持首次云部署的简单性。一旦同步调用正常运作，开发者就可以在减少未知变量的情况下部署串流或大容量 Payload 变体。

● 使用环境变量存储 ARN： 将运行时 ARN 导出为环境变量，可以将部署与调用脚本解耦。这模拟了生产环境模式，其中部署输出会提供给客户端设置或 CI/CD 变量。

---

步骤 6 — 使用 boto3 调用运行时

开发者操作

建立 invoke_runtime.py。

import json
import os
import uuid
import boto3
from validate_payload import validate_payload

region = os.getenv("AWS_DEFAULT_REGION", "us-east-1")
agent_arn = os.environ["AGENTCORE_RUNTIME_ARN"]
session_id = os.getenv("RUNTIME_SESSION_ID", str(uuid.uuid4()))

payload = {
    "request_id": "hands-on-001",
    "prompt": (
        "分析美國 10 年期 4.25 對比中國 10 年期 2.15。"
        "USD 風險敞口為 25000000 且避險比率為 0.60。"
        "部位名目本金為 10000000，價差變動為 25 bps，存續期間為 5。"
        "請包含確認訊號與失效觸發條件。"
    ),
}

ok, errors = validate_payload(payload)
if not ok:
    raise ValueError(errors)

client = boto3.client("bedrock-agentcore", region_name=region)
res = client.invoke_agent_runtime(
    agentRuntimeArn=agent_arn,
    runtimeSessionId=session_id,
    qualifier="DEFAULT",
    payload=json.dumps(payload).encode("utf-8"),
)

body = b"".join(res["response"]).decode("utf-8")
print(body)
print("工作階段 ID:", session_id)

执行：

python invoke_runtime.py

商业逻辑

客户端请求结构化的主权风险分析，并提供应触发计算工具的数值。

代码逻辑

脚本验证 Payload、建立 boto3 AgentCore 客户端、传送 JSON 位元组、组合运行时响应区块（Chunks），并印出会话 ID。

预期结果

响应包含分析内容以及计算出的数值，例如基点价差、避险名目本金或存续期间影响。

系统设计决策

● 客户端验证： 客户端在传送请求前进行验证。这可以及早发现错误，并避免不必要的运行时调用。生产环境的客户端可以共用相同的验证逻辑或使用正式的 JSON Schema。

● 会话 ID 重复使用： 脚本会印出会话 ID，以便开发者可以将其重复用于多轮（Multi-turn）测试。这展示了 AgentCore 会话如何支持跨调用的上下文工作流程。

● 响应正规化： 脚本将响应区块组合成单一字符串。这种抽象化保持了下游客户端的简单性，同时仍能与事件风格的运行时响应兼容。

---

步骤 7 — 加入串流运行时变体

开发者操作

建立 app_streaming.py。

from strands import Agent, tool
from strands.models import BedrockModel
from bedrock_agentcore.runtime import BedrockAgentCoreApp

app = BedrockAgentCoreApp()

@tool
def credit_spread_widening() -> str:
    """回傳信用利差擴大的核心驅動因素。"""
    return "驅動因素包括資金壓力、流動性撤出、調降評級風險以及防禦性部位輪動。"

agent = Agent(
    model=BedrockModel(model_id="amazon.nova-pro-v1:0", temperature=0.2, max_tokens=4000),
    tools=[credit_spread_widening],
    system_prompt="串流輸出簡明的主權風險分析，包含證據、確認與失效章節。",
)

@app.entrypoint
async def stream_runtime(payload, context):
    prompt = payload.get("prompt", "")
    if not prompt:
        yield {"type": "error", "message": "缺少 prompt"}
        return

    request = f"工作階段: {context.session_id}\n{prompt}"
    async for event in agent.stream_async(request):
        if "data" in event:
            yield event["data"]

if __name__ == "__main__":
    app.run()

商业逻辑

串流功能允许客户端在分析内容生成时同步显示，这对分析师仪表板和对话界面非常有用。

代码逻辑

进入点为 async（非同步），调用 agent.stream_async()，并产出（Yield）数据区块。

预期结果

支持串流的运行时可以返回部分区块，而不需要等待完整响应生成完毕。

系统设计决策

● 用于部分输出的非同步进入点： 串流需要一个能产出区块的非同步进入点。这改善了用户界面的感知延迟（Perceived latency），并向开发者展示了不同的运行时响应模式。

● 独立的串流文件： 将串流代码与同步运行时分开，可以轻松进行对比。开发者可以清楚地看到改变了什么：改变的是进入点风格与响应处理方式，而非整个架构。

● 工具支持依然可用： 串流代理人仍然保有工具。这证明了串流是一种传输选择，并不会削弱代理人的能力。

---

步骤 8 — 加入大容量 Payload 处理器

开发者操作

建立 app_large_payload.py。

import base64
from strands import Agent
from strands.models import BedrockModel
from bedrock_agentcore.runtime import BedrockAgentCoreApp

app = BedrockAgentCoreApp()

agent = Agent(
    model=BedrockModel(model_id="amazon.nova-pro-v1:0", temperature=0.2, max_tokens=8000),
    system_prompt=(
        "分析提供的財務文件與圖表以評估主權風險、殖利率價差、信用利差、"
        "外匯壓力、流動性、資本流動、避險考量、確認訊號與失效觸發條件。"
        "請勿提供投資建議。"
    ),
)

@app.entrypoint
def large_payload_runtime(payload, context):
    content = [{"text": f"工作階段: {context.session_id}\n{payload.get('prompt', '分析提供的資料')}"}]

    if payload.get("excel_data"):
        content.insert(0, {
            "document": {
                "format": "xlsx",
                "name": "sovereign_dataset",
                "source": {"bytes": base64.b64decode(payload["excel_data"])},
            }
        })

    if payload.get("image_data"):
        content.insert(0, {
            "image": {
                "format": "png",
                "source": {"bytes": base64.b64decode(payload["image_data"])},
            }
        })

    response = agent(content)
    return response.message["content"][0]["text"]

商业逻辑

运行时将用户提示词、试算表数据和图表影像信号整合到单一分析流程中。

代码逻辑

处理器对 base64 Payload 字段进行解码，并构建具备类型的 Bedrock 内容区块：document、image 和 text。

预期结果

客户端可以传送选填的 excel_data 和 image_data 字段，并接收整合后的分析结果。

系统设计决策

● 用于 JSON 传输的 Base64： 二进位文件经过编码，以便通过 JSON Payload 传输。这保持了客户端契约的简单性，并避免了工作坊期间多部分上传（Multipart upload）的复杂性。

● 具备类型的内容区块： 传入 document 和 image 区块可以保留模态性（Modality）。这比将所有内容转换为文字更具维护性，并能让模型利用文件/影像特定的专属能力。

● 选填的 Payload 字段： 处理器可与纯文字、纯文件、纯影像或组合请求协同运作。这使得运行时对于多种客户端类型更具弹性。

---

步骤 9 — 加入显式会话清理

# stop_session.py
import os
import boto3

client = boto3.client("bedrock-agentcore", region_name=os.getenv("AWS_DEFAULT_REGION", "us-east-1"))
client.stop_runtime_session(
    agentRuntimeArn=os.environ["AGENTCORE_RUNTIME_ARN"],
    runtimeSessionId=os.environ["RUNTIME_SESSION_ID"],
    qualifier="DEFAULT",
)
print("已停止工作階段", os.environ["RUNTIME_SESSION_ID"])

系统设计决策

● 会话是受管资源： 代理人会话可以保留上下文并消耗资源。开发者应该在工作流程结束时显式停止会话，而不是仅依赖闲置过期。

● 清理脚本化： 一个小脚本很容易在展示、测试和 CI 工作结束后执行。这鼓励了规范的运行时维护习惯。

● 运作控制把手： 会话 ID 成为调试、追踪与清理的运作控制把手（Operational handle）。

---

开发者最终检查清单

● [ ] pytest -q 在本地通过。

● [ ] python deploy_runtime.py 返回运行时 ARN。

● [ ] python invoke_runtime.py 返回结构化分析。

● [ ] 串流进入点编译成功且可单独部署。

● [ ] 大容量 Payload 进入点接受选填的 excel_data 和 image_data。

● [ ] 会话清理脚本已通过测试。

---

额外开发者实战实验室

以下实验室扩展了运行时工作坊，提供了更深入的开发者实施练习。它们被设计为核心两小时构建后的选修模块，或适合希望加强部署、测试和运营规范的专业团队的后续练习。

---

实战实验室 A — 为运行时 Payload 新增 JSON Schema 验证

开发者目标

将简单的字段验证器替换为可重复使用的 JSON Schema 验证器，该验证器可由客户端、测试和运行时处理器共用。

开发者操作

安装 jsonschema：

pip install jsonschema

建立 payload_schema.py：

RUNTIME_PAYLOAD_SCHEMA = {
    "type": "object",
    "properties": {
        "request_id": {"type": "string", "minLength": 1},
        "user_id": {"type": "string", "minLength": 1},
        "prompt": {"type": "string", "minLength": 1},
        "excel_data": {"type": "string"},
        "image_data": {"type": "string"},
        "metadata": {
            "type": "object",
            "additionalProperties": {"type": ["string", "number", "boolean", "null"]},
        },
    },
    "required": ["prompt"],
    "additionalProperties": False,
}

建立 validate_schema.py：

from jsonschema import Draft202012Validator
from payload_schema import RUNTIME_PAYLOAD_SCHEMA

validator = Draft202012Validator(RUNTIME_PAYLOAD_SCHEMA)

def validate_payload_schema(payload: dict) -> tuple[bool, list[str]]:
    errors = sorted(validator.iter_errors(payload), key=lambda e: e.path)
    messages = [f"{list(error.path)}: {error.message}" for error in errors]
    return len(messages) == 0, messages

if __name__ == "__main__":
    sample = {"prompt": "分析價差", "metadata": {"source": "lab"}}
    ok, messages = validate_payload_schema(sample)
    print(ok)
    print(messages)

商业逻辑

Schema 正式确立了运行时 API 契约。这有助于前端开发者、后端服务和测试软件包在运行时支持的确切字段上达成共识。

代码逻辑

Draft202012Validator 检查 Payload 的形状、必要字段、数据类型与未知字段。辅助程序返回一个布尔值与人类可读的错误消息。

预期结果

python validate_schema.py
# True
# []

系统设计决策

● Schema 作为共享 API 契约： JSON Schema 比手写验证更精确，因为它在一个可重复使用的构件中定义了类型、必要字段与未知字段的行为。这允许客户端、测试软件包和运行时处理器在调用 AgentCore Runtime 之前验证相同的契约。

● 在模型调用前快速失败： 运行时调用可能会涉及模型延迟与成本。在调用代理人之前验证 Payload 可以防止本可避免的失败，并为开发者提供实时反馈。它还减少了由格式错误的客户端 Payload 引起的异常运行时日志。

● 可扩展的元数据对象： Schema 允许使用 metadata 对象来提供安全的运营上下文，同时封锁任意的最上层字段。这使得团队在保持弹性的同时，不会让运行时契约变得不受控制。

---

实战实验室 B — 构建可重复使用的运行时客户端类别

开发者目标

将 boto3 调用、会话 ID、Payload 验证和响应正规化封装到一个可重复使用的客户端类别中。

开发者操作

建立 runtime_client.py：

import json
import os
import uuid
import boto3
from validate_schema import validate_payload_schema

class AgentCoreRuntimeClient:
    def __init__(self, runtime_arn: str, region: str | None = None):
        self.runtime_arn = runtime_arn
        self.region = region or os.getenv("AWS_DEFAULT_REGION", "us-east-1")
        self.client = boto3.client("bedrock-agentcore", region_name=self.region)

    def invoke(self, prompt: str, session_id: str | None = None, **kwargs) -> dict:
        payload = {"prompt": prompt, **kwargs}
        ok, errors = validate_payload_schema(payload)
        if not ok:
            raise ValueError({"payload_errors": errors})

        runtime_session_id = session_id or str(uuid.uuid4())
        response = self.client.invoke_agent_runtime(
            agentRuntimeArn=self.runtime_arn,
            runtimeSessionId=runtime_session_id,
            qualifier="DEFAULT",
            payload=json.dumps(payload).encode("utf-8"),
        )
        body = b"".join(response["response"]).decode("utf-8")
        return {"session_id": runtime_session_id, "body": body}

建立 use_runtime_client.py：

import os
from runtime_client import AgentCoreRuntimeClient

client = AgentCoreRuntimeClient(os.environ["AGENTCORE_RUNTIME_ARN"])
result = client.invoke(
    prompt="分析美中殖利率價差變動，並包含確認與失效訊號。",
    request_id="client-lab-001",
    metadata={"application": "developer-lab"},
)
print(result["session_id"])
print(result["body"])

商业逻辑

应用程序团队需要一个乾净的运行时整合层，而不是在每个服务中重复编写 boto3 代码。

代码逻辑

该类别验证 Payload、建立或重复使用会话 ID、调用 AgentCore Runtime、组合响应事件，并返回正规化的输出。

预期结果

开发者只需两行应用程序代码即可调用运行时。

系统设计决策

● 客户端抽象化： 可重复使用的类别可以防止在各个应用程序中重复编写 boto3 样板代码。它还集中了验证、响应正规化和会话处理，从而使整合行为保持一致。

● 设计上的会话连续性： 客户端允许调用者传入会话 ID 或建立一个新的会话 ID。这使得多轮工作流程变得明确，同时保留了简单的单次（One-shot）调用。

● 正规化的返回形状： 返回 {session_id, body} 使应用程序代码更容易测试，并避免了将底层 boto3 事件详细信息泄露给每个调用者。

---

实战实验室 C — 加入运行时错误分类法

开发者目标

为运行时客户端和运营人员提供一致的错误类型，以应对缺少字段、模型失败和非预期异常。

开发者操作

建立 errors.py：

class RuntimeErrorCode:
    MISSING_PROMPT = "MISSING_PROMPT"
    VALIDATION_ERROR = "VALIDATION_ERROR"
    AGENT_FAILURE = "AGENT_FAILURE"
    UNEXPECTED_ERROR = "UNEXPECTED_ERROR"

def error_response(code: str, message: str, request_id: str | None = None) -> dict:
    return {
        "status": "error",
        "error": {
            "code": code,
            "message": message,
            "request_id": request_id,
        },
    }

修改运行时进入点模式：

from errors import RuntimeErrorCode, error_response

@app.entrypoint
def sovereign_runtime(payload, context):
    request_id = payload.get("request_id", context.session_id)
    try:
        prompt = payload.get("prompt", "")
        if not prompt.strip():
            return error_response(RuntimeErrorCode.MISSING_PROMPT, "提示詞 (prompt) 為必要欄位", request_id)
        response = agent(f"請求 ID: {request_id}\n工作階段: {context.session_id}\n{prompt}")
        return {"status": "ok", "request_id": request_id, "response": response.message["content"][0]["text"]}
    except Exception as exc:
        return error_response(RuntimeErrorCode.UNEXPECTED_ERROR, str(exc), request_id)

商业逻辑

运营系统需要可预测的错误，以便安全地进行路由、告警、重试或显示。

代码逻辑

错误辅助程序返回一个包含状态、错误码、消息和请求 ID 的一致对象。

预期结果

格式错误的请求会返回结构化错误，而不是不一致的字符串或堆叠追踪（Stack traces）。

系统设计决策

● 用于运营的错误分类法： 生产环境的客户端需要区分验证失败与代理人失败。基于代码的分类法允许仪表板、重试逻辑和告警规则做出相应的反应。

● 每个错误中都包含请求 ID： 在错误中包含请求 ID 有助于将客户端失败与运行时日志和会话 ID 相关联。这减少了工作坊与生产事件期间的调试时间。

● 安全的异常处理： 运行时提取非预期错误并返回一个受控的响应。这可以防止原始堆叠追踪泄露给调用者，同时仍为开发者调试保留足够的详细信息。

---

实战实验室 D — 构建串流 CLI 消费者

开发者目标

建立一个客户端，可以解析服务器传送事件（Server-Sent Event, SSE）风格的串流响应并逐步印出区块。

开发者操作

建立 invoke_streaming_runtime.py：

import json
import os
import uuid
import boto3

client = boto3.client("bedrock-agentcore", region_name=os.getenv("AWS_DEFAULT_REGION", "us-east-1"))
session_id = str(uuid.uuid4())

response = client.invoke_agent_runtime(
    agentRuntimeArn=os.environ["AGENTCORE_STREAMING_RUNTIME_ARN"],
    runtimeSessionId=session_id,
    qualifier="DEFAULT",
    payload=json.dumps({
        "prompt": "串流分析信用利差擴大與主權重新定價訊號。"
    }).encode("utf-8"),
)

print("工作階段:", session_id)
for event in response["response"]:
    chunk = event.decode("utf-8") if isinstance(event, bytes) else str(event)
    print(chunk, end="", flush=True)
print()

商业逻辑

串流客户端允许开发者查看渐进式的模型输出，这对于聊天 UI 和分析师仪表板非常有用。

代码逻辑

脚本调用串流运行时，并在响应事件到达时对其进行反覆运算。

预期结果

终端机会渐进式地印出内容，而不是等待完整响应。

系统设计决策

● 渐进式转译客户端： 只有当客户端正确使用区块时，串流功能才有帮助。本实验室教授的是串流的调用端，而不仅仅是运行时端。

● 独立的串流运行时 ARN： 脚本使用独立的环境变量，以避免混淆同步与串流部署。这保持了测试的明确性。

● 立即排空输出： flush=True 模拟了 UI 的渐进式转译，并有助于开发者在终端机中观察串流行为。

---

实战实验室 E — 加入本地成本与延迟日志包装器

开发者目标

在应用程序层提取模型调用延迟，以便进行工作坊调试和未来的可观测性构建。

开发者操作

建立 timed_agent.py：

import time
import json
from datetime import datetime, timezone

class TimedAgent:
    def __init__(self, agent, name: str):
        self.agent = agent
        self.name = name

    def __call__(self, prompt):
        start = time.time()
        response = self.agent(prompt)
        elapsed_ms = int((time.time() - start) * 1000)
        print(json.dumps({
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "event_type": "agent_invocation_latency",
            "agent": self.name,
            "elapsed_ms": elapsed_ms,
        }))
        return response

包装代理人：

from timed_agent import TimedAgent
agent = TimedAgent(agent, "sovereign-runtime-agent")

商业逻辑

在新增受管的可观测性仪表板之前，开发者需要对缓慢的模型调用和运行时行为具备可见性。

代码逻辑

TimedAgent 装饰（Decorate）一个现有的 Strands 代理人，并为每次调用记录耗费的毫秒数。

预期结果

每次调用都会印出一个 JSON 延迟事件。

系统设计决策

● 使用装饰器而非侵入式修改： 包装代理人可以避免更改代理人的构建或商业逻辑。这保持了可观测性的模块化，且易于移除或替换。

● 结构化延迟日志： JSON 延迟日志可以被搜索与汇整。它们还建立了与 AgentCore Observability 和 CloudWatch 的自然桥梁。

● 开发者反馈循环： 延迟数据有助于开发者在实战实验期间，了解较长提示词、较大 Payload 以及额外工具所带来的性能成本。