Build with AgentCore & Strands: Runtime Sovereign Risk Agent Developer Workshop

Audience: professional Python/AWS developers Duration: 2 hours Primary AWS AI services: Amazon Bedrock AgentCore Runtime, Strands Agents, Amazon Bedrock models Workshop style: hands-on developer build, no slides required Project output: a working runtime agent with local validation, deployment script, boto3 invocation client, streaming variant, session cleanup, and large-payload handler.

Educational engineering workshop only. The sample domain is sovereign-risk analytics, but the output is not financial, investment, legal, tax, or trading advice.

Workshop Summary

Developers build a production-style sovereign-risk agent using Strands and AgentCore Runtime. The workshop covers deterministic calculation tools, payload validation, smoke tests, deployment, boto3 invocation, streaming responses, large base64 payload handling, and session cleanup. Participants leave with reusable runtime patterns for secure, observable, session-aware AI services that combine model reasoning with testable Python logic on AWS for modern enterprise developer workflows.

---

1. Developer learning objectives

By the end of this workshop, developers can:

● Build a Strands agent backed by an Amazon Bedrock model.

● Add deterministic Python tools to an LLM-driven agent.

● Wrap the agent in an Amazon Bedrock AgentCore Runtime entrypoint.

● Package and launch the runtime using the AgentCore starter toolkit.

● Invoke the runtime with boto3 using a stable session ID.

● Implement a streaming runtime entrypoint for responsive clients.

● Build a large-payload entrypoint for base64-encoded Excel and image inputs.

● Add local payload validation and smoke tests before deployment.

---

2. Architecture built in this workshop

Developer laptop
  ├─ app.py                         # synchronous AgentCore Runtime entrypoint
  ├─ app_streaming.py               # streaming Runtime entrypoint
  ├─ app_large_payload.py           # multimodal / large-payload Runtime entrypoint
  ├─ deploy_runtime.py              # packages and launches runtime
  ├─ invoke_runtime.py              # boto3 client invocation
  ├─ stop_session.py                # explicit session cleanup
  ├─ validate_payload.py            # local payload contract validation
  ├─ smoke_test.py                  # local smoke checks
  ├─ prompts/system.md              # governed system prompt
  └─ requirements.txt

AWS
  ├─ Amazon Bedrock model through Strands
  ├─ Amazon Bedrock AgentCore Runtime
  ├─ Amazon ECR image created by toolkit
  ├─ IAM execution role
  └─ Cloud logs / runtime output

---

3. 2-hour hands-on agenda

---

4. Prerequisites

python --version     # recommended: 3.11+
aws --version
aws sts get-caller-identity
export AWS_DEFAULT_REGION=us-east-1

Create and activate a virtual environment:

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

---

Step 1 — Create the project scaffold

Developer action

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

Create the system prompt:

cat > prompts/system.md <<'EOF'
You are a professional sovereign risk engineering assistant.
Analyze yield spreads, credit spread widening, sovereign debt repricing, FX pressure,
funding liquidity, duration demand, capital flows, and defensive rotation.
Always return these sections:
1. Observation
2. Reasoning
3. Risk implication
4. Hedge consideration
5. Confirmation signals
6. Invalidation triggers
7. Limitations
Use tools for numeric calculations. Do not provide investment advice or autonomous trading orders.
EOF

Install dependencies:

pip install -r requirements.txt

System design decision

● Prompt file as source-controlled behavior: The system prompt is part of the runtime contract. Keeping it in prompts/system.md lets developers review behavior changes like code changes. This matters because agent output style, safety boundaries, and required sections affect downstream clients, tests, and evaluations.

● Explicit dependency file: A runtime deployment must be reproducible. requirements.txt defines the exact Python package surface the runtime expects. It also makes deployment packaging, container builds, and workshop troubleshooting easier because every participant installs the same dependency set.

● Small scaffold before cloud deployment: Developers first create local files and validate behavior before deploying. This reduces cloud debugging noise. If a prompt file is missing or a Python import fails, the error is caught locally before the runtime is packaged and launched.

Expected result

The project contains a dependency file and a governed system prompt that will be read by the Strands agent.

---

Step 2 — Build the synchronous Strands agent runtime

Developer action

Create 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:
    """Calculate the spread between two yields in basis points."""
    spread = (us_yield - peer_yield) * 100
    return f"Yield spread is {spread:.1f} basis points."

@tool
def fx_hedge_notional(exposure_usd: float, hedge_ratio: float) -> str:
    """Calculate hedge notional from USD exposure and hedge ratio."""
    hedge = exposure_usd * hedge_ratio
    return f"FX hedge notional is USD {hedge:,.2f}."

@tool
def stress_loss_notional(position_notional: float, spread_move_bps: float, duration: float) -> str:
    """Estimate duration-driven price impact from spread move."""
    loss = position_notional * duration * (spread_move_bps / 10000)
    return f"Estimated duration impact is 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": "Missing required field: prompt"}

    request_id = payload.get("request_id", context.session_id)
    request = (
        f"Request ID: {request_id}\n"
        f"Runtime session: {context.session_id}\n"
        f"User request: {prompt}"
    )
    response = agent(request)
    return response.message["content"][0]["text"]

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

Business logic

The agent analyzes sovereign-risk prompts and uses tools for basis-point spreads, FX hedge notional, and duration stress impact. The model handles qualitative synthesis, while deterministic calculations stay in Python.

Code logic

● BedrockAgentCoreApp() creates the runtime application wrapper.

● BedrockModel() configures the Amazon Bedrock model used by Strands.

● @tool exposes Python functions to the Strands agent.

● @app.entrypoint marks the callable runtime handler.

● The handler validates prompt, adds request/session context, invokes the agent, and returns the final text.

Expected result

Running python app.py starts a local AgentCore-compatible runtime process. When deployed, the same entrypoint receives JSON payloads from AgentCore Runtime.

System design decision

● Deterministic tools for quantitative logic: LLMs are useful for synthesis, but financial calculations should be deterministic and testable. By moving yield spread, hedge notional, and duration stress math into Python tools, developers get repeatable outputs and can unit-test calculations independently from model behavior.

● Request ID and session context: The entrypoint injects request and session IDs into the model prompt. This gives responses operational context and helps engineers correlate runtime logs, client calls, and user workflows. It also creates a clean pattern for production tracing.

● Low-temperature Bedrock model: The agent uses temperature=0.2 to reduce response variance. Professional developer workshops should teach stable engineering patterns, not creative prompt experimentation. Lower variance improves smoke testing, demos, and future evaluation baselines.

---

Step 3 — Add local payload validation

Developer action

Create 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"Missing required field: {field}")

    unknown = set(payload.keys()) - set(REQUIRED_FIELDS) - set(OPTIONAL_FIELDS)
    for field in sorted(unknown):
        errors.append(f"Unknown field: {field}")

    return len(errors) == 0, errors

if __name__ == "__main__":
    sample = {"prompt": "Analyze US-China yield spread", "request_id": "demo-001"}
    ok, errors = validate_payload(sample)
    print("valid", ok)
    print("errors", errors)

Business logic

The validator enforces the request contract before runtime invocation. This prevents clients from sending malformed requests and makes debugging easier.

Code logic

The function checks required fields, rejects unknown fields, and returns a boolean plus error list. It can be imported by tests, clients, or runtime code.

Expected result

python validate_payload.py
# valid True
# errors []

System design decision

● Payload contract before deployment: Developers should validate API contracts locally before invoking cloud runtime. This reduces failed runtime calls caused by missing prompts or misspelled fields. It also creates a foundation for stronger schemas later.

● Explicit unknown-field detection: Rejecting unknown fields helps catch integration mistakes early. Without this, clients can believe metadata is being used when the runtime silently ignores it. Clear validation improves developer feedback.

● Reusable validation module: The validator is a module, not only a script. It can be reused in unit tests, CLI clients, and the runtime entrypoint. This avoids duplicating contract logic in multiple files.

---

Step 4 — Add smoke tests

Developer action

Create 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": "Analyze spread"})
    assert ok
    assert errors == []

def test_missing_prompt_fails():
    ok, errors = validate_payload({"request_id": "x"})
    assert not ok
    assert "Missing required field: 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)

Run:

pytest -q

Business logic

Smoke tests verify the request contract and deterministic calculation tools before deploying the agent.

Code logic

The tests import validation and tool functions directly. They do not invoke the model, which keeps tests fast and deterministic.

Expected result

3 passed

System design decision

● Test deterministic parts first: Model output can vary, but tool math and payload validation should not. Testing deterministic components gives developers fast feedback and isolates bugs before cloud deployment.

● No model dependency in smoke tests: The smoke tests do not call Amazon Bedrock. This avoids cost, credentials, model latency, and nondeterministic assertions. The goal is to validate local software contracts.

● Workshop-friendly failure messages: Simple assertions make it clear what failed. In a two-hour workshop, debugging must be fast and focused. These tests catch common setup and logic mistakes early.

---

Step 5 — Deploy AgentCore Runtime

Developer action

Create 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)

Run:

python deploy_runtime.py
export AGENTCORE_RUNTIME_ARN="paste-agent-runtime-arn"

Business logic

The local agent becomes a managed runtime endpoint that applications can invoke.

Code logic

Runtime().configure() packages the entrypoint and dependencies. launch() deploys the runtime and returns identifiers.

Expected result

The script prints an AgentCore Runtime ARN and runtime ID.

System design decision

● Toolkit for hands-on speed: The starter toolkit handles deployment mechanics so the workshop can focus on runtime design, not container internals. Developers still see the runtime ARN and can inspect generated AWS resources afterward.

● Single runtime entrypoint first: Deploying only app.py keeps the first cloud deployment simple. Once synchronous invocation works, developers can deploy streaming or large-payload variants with fewer unknowns.

● Environment variable for ARN: Exporting the runtime ARN decouples deployment from invocation scripts. This mirrors production patterns where deployment outputs feed client configuration or CI/CD variables.

---

Step 6 — Invoke runtime with boto3

Developer action

Create 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": (
        "Analyze US 10Y 4.25 vs China 10Y 2.15. "
        "USD exposure is 25000000 and hedge ratio is 0.60. "
        "Position notional is 10000000, spread move is 25 bps, duration is 5. "
        "Include confirmation signals and invalidation triggers."
    ),
}

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("Session ID:", session_id)

Run:

python invoke_runtime.py

Business logic

The client asks for a structured sovereign-risk analysis and provides numeric values that should trigger calculation tools.

Code logic

The script validates the payload, creates a boto3 AgentCore client, sends JSON bytes, joins runtime response chunks, and prints the session ID.

Expected result

The response includes analysis plus calculated values such as basis-point spread, hedge notional, or duration impact.

System design decision

● Client-side validation: The client validates before sending requests. This catches mistakes earlier and avoids unnecessary runtime invocations. Production clients can share the same validation logic or use a formal JSON Schema.

● Session ID reuse: The script prints the session ID so developers can reuse it for multi-turn testing. This demonstrates how AgentCore sessions support contextual workflows across calls.

● Response normalization: The script joins response chunks into a single string. This abstraction keeps downstream clients simple while still being compatible with event-style runtime responses.

---

Step 7 — Add streaming runtime variant

Developer action

Create 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 core drivers of credit spread widening."""
    return "Drivers include funding stress, liquidity withdrawal, downgrade risk, and defensive positioning."

agent = Agent(
    model=BedrockModel(model_id="amazon.nova-pro-v1:0", temperature=0.2, max_tokens=4000),
    tools=[credit_spread_widening],
    system_prompt="Stream concise sovereign-risk analysis with evidence, confirmation, and invalidation sections.",
)

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

    request = f"Session: {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()

Business logic

Streaming lets a client display analysis as it is generated, useful for analyst dashboards and chat interfaces.

Code logic

The entrypoint is async, calls agent.stream_async(), and yields data chunks.

Expected result

A streaming-capable runtime can return partial chunks instead of waiting for the complete response.

System design decision

● Async entrypoint for partial output: Streaming requires an asynchronous entrypoint that yields chunks. This improves perceived latency for user interfaces and teaches developers a different runtime response pattern.

● Separate streaming file: Keeping streaming code separate from the synchronous runtime makes comparison easy. Developers can see exactly what changes: entrypoint style and response handling, not the entire architecture.

● Tool support remains available: The streaming agent still has tools. This proves that streaming is a transport choice, not a reduction in agent capability.

---

Step 8 — Add large-payload handler

Developer action

Create 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=(
        "Analyze financial documents and charts for sovereign risk, yield spreads, credit spreads, "
        "FX pressure, liquidity, capital flows, hedge considerations, confirmation signals, and invalidation triggers. "
        "Do not provide investment advice."
    ),
)

@app.entrypoint
def large_payload_runtime(payload, context):
    content = [{"text": f"Session: {context.session_id}\n{payload.get('prompt', 'Analyze provided data')}"}]

    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"]

Business logic

The runtime combines user prompt, spreadsheet data, and chart image signals into one analysis flow.

Code logic

The handler decodes base64 payload fields and builds typed Bedrock content blocks: document, image, and text.

Expected result

A client can send optional excel_data and image_data fields and receive a combined analysis.

System design decision

● Base64 for JSON transport: Binary files are encoded so they can travel through JSON payloads. This keeps the client contract simple and avoids multipart upload complexity during the workshop.

● Typed content blocks: Passing document and image blocks preserves modality. This is more maintainable than converting everything to text and lets the model use document/image-specific capabilities.

● Optional payload fields: The handler works with text-only, document-only, image-only, or combined requests. This makes the runtime flexible for multiple client types.

---

Step 9 — Add explicit session cleanup

# 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("Stopped session", os.environ["RUNTIME_SESSION_ID"])

System design decision

● Sessions are managed resources: Agent sessions can preserve context and consume resources. Developers should explicitly stop sessions when workflows end rather than relying only on idle expiration.

● Cleanup as script: A small script is easy to run after demos, tests, and CI jobs. This encourages disciplined runtime hygiene.

● Operational handle: The session ID becomes an operational handle for debugging, tracing, and cleanup.

---

Final developer checklist

● [ ] pytest -q passes locally.

● [ ] python deploy_runtime.py returns runtime ARN.

● [ ] python invoke_runtime.py returns structured analysis.

● [ ] Streaming entrypoint compiles and can be deployed separately.

● [ ] Large-payload entrypoint accepts optional excel_data and image_data.

● [ ] Session cleanup script is tested.

---

Additional Hands-on Developer Labs

The following labs extend the runtime workshop with deeper developer practice. They are designed as optional modules after the core two-hour build or as follow-up exercises for professional teams that want stronger deployment, testing, and operations discipline.

---

Hands-on Lab A — Add JSON Schema validation for runtime payloads

Developer goal

Replace the simple field validator with a reusable JSON Schema validator that can be shared by clients, tests, and runtime handlers.

Developer action

Install jsonschema:

pip install jsonschema

Create 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,
}

Create 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": "Analyze spread", "metadata": {"source": "lab"}}
    ok, messages = validate_payload_schema(sample)
    print(ok)
    print(messages)

Business logic

The schema formalizes the runtime API contract. This helps front-end developers, backend services, and test suites agree on the exact fields the runtime supports.

Code logic

Draft202012Validator checks payload shape, required fields, data types, and unknown fields. The helper returns a boolean plus human-readable errors.

Expected result

python validate_schema.py
# True
# []

System design decision

● Schema as shared API contract: A JSON Schema is more precise than hand-written validation because it defines types, required fields, and unknown-field behavior in one reusable artifact. This allows clients, test suites, and runtime handlers to validate the same contract before calling AgentCore Runtime.

● Fail fast before model invocation: Runtime calls may involve model latency and cost. Validating payloads before invoking the agent prevents avoidable failures and gives developers immediate feedback. It also reduces noisy runtime logs caused by malformed client payloads.

● Extensible metadata object: The schema allows a metadata object for safe operational context while blocking arbitrary top-level fields. This gives teams flexibility without letting the runtime contract become uncontrolled.

---

Hands-on Lab B — Build a reusable runtime client class

Developer goal

Encapsulate boto3 invocation, session IDs, payload validation, and response normalization into one reusable client.

Developer action

Create 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}

Create use_runtime_client.py:

import os
from runtime_client import AgentCoreRuntimeClient

client = AgentCoreRuntimeClient(os.environ["AGENTCORE_RUNTIME_ARN"])
result = client.invoke(
    prompt="Analyze US-China yield spread movement with confirmation and invalidation signals.",
    request_id="client-lab-001",
    metadata={"application": "developer-lab"},
)
print(result["session_id"])
print(result["body"])

Business logic

Application teams need a clean runtime integration layer instead of duplicating boto3 code in every service.

Code logic

The class validates payloads, creates or reuses a session ID, invokes AgentCore Runtime, joins response events, and returns normalized output.

Expected result

Developers can call the runtime with two lines of application code.

System design decision

● Client abstraction: A reusable class prevents repeated boto3 boilerplate across applications. It also centralizes validation, response normalization, and session handling so integration behavior remains consistent.

● Session continuity by design: The client lets callers pass a session ID or create a new one. This makes multi-turn workflows explicit while preserving simple one-shot invocation.

● Normalized return shape: Returning {session_id, body} makes application code easier to test and avoids leaking low-level boto3 event details into every caller.

---

Hands-on Lab C — Add runtime error taxonomy

Developer goal

Give runtime clients and operators consistent error types for missing fields, model failures, and unexpected exceptions.

Developer action

Create 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,
        },
    }

Modify the runtime entrypoint pattern:

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 is required", request_id)
        response = agent(f"Request ID: {request_id}\nSession: {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)

Business logic

Operational systems need predictable errors that can be routed, alerted, retried, or displayed safely.

Code logic

The error helper returns a consistent object with status, code, message, and request ID.

Expected result

Malformed requests return structured errors instead of inconsistent strings or stack traces.

System design decision

● Error taxonomy for operations: Production clients need to distinguish validation failures from agent failures. A code-based taxonomy allows dashboards, retry logic, and alerting rules to react appropriately.

● Request ID in every error: Including request ID in errors helps correlate client failures with runtime logs and session IDs. This reduces debugging time during workshops and production incidents.

● Safe exception handling: The runtime catches unexpected errors and returns a bounded response. This prevents raw stack traces from leaking to callers while still preserving enough detail for developer debugging.

---

Hands-on Lab D — Build a streaming CLI consumer

Developer goal

Create a client that can parse Server-Sent Event style streaming responses and print chunks progressively.

Developer action

Create 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": "Stream analysis of credit spread widening and sovereign repricing signals."
    }).encode("utf-8"),
)

print("Session:", 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()

Business logic

A streaming client lets developers see incremental model output, which is useful for chat UIs and analyst dashboards.

Code logic

The script invokes the streaming runtime and iterates over response events as they arrive.

Expected result

The terminal prints content progressively instead of waiting for the full response.

System design decision

● Progressive rendering client: Streaming only helps if clients consume chunks correctly. This lab teaches the caller side of streaming, not only the runtime side.

● Separate streaming runtime ARN: The script uses a distinct environment variable to avoid confusing synchronous and streaming deployments. This keeps testing explicit.

● Flush output immediately: flush=True simulates UI progressive rendering and helps developers observe streaming behavior in a terminal.

---

Hands-on Lab E — Add local cost and latency logging wrapper

Developer goal

Capture model invocation latency at the application layer for workshop debugging and future observability.

Developer action

Create 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

Wrap the agent:

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

Business logic

Developers need visibility into slow model calls and runtime behavior before adding managed observability dashboards.

Code logic

TimedAgent decorates an existing Strands agent and logs elapsed milliseconds for every invocation.

Expected result

Each invocation prints a JSON latency event.

System design decision

● Decorator instead of invasive changes: Wrapping the agent avoids changing agent construction or business logic. This keeps observability modular and easy to remove or replace.

● Structured latency logs: JSON latency logs can be searched and aggregated. They also create a natural bridge to AgentCore Observability and CloudWatch.

● Developer feedback loop: Latency data helps developers understand the cost of longer prompts, larger payloads, and additional tools during hands-on experimentation.