Build with AgentCore & Strands: Governed Multi-Agent Photolithography Drift Automation Developer Workshop

Audience: senior developers, platform engineers, AI governance engineers Duration: 2 hours Primary AWS AI services: Amazon Bedrock AgentCore Runtime, AgentCore Memory pattern, AgentCore Policy pattern, AgentCore Observability pattern, AgentCore Evaluations pattern, Strands Agents Project output: a governed multi-agent runtime with specialist Strands agents, bounded memory, policy checks, structured telemetry, evaluation fixtures, and a runtime client.

Educational engineering workshop only. This is a software architecture exercise and not process-release advice.

This workshop teaches developers to design a governed multi-agent automation system with AgentCore Runtime and Strands specialists. Participants implement orchestration, request and response policy checks, bounded memory summaries, structured observability events, and evaluation fixtures for safe and blocked scenarios. The result is a reusable architecture for evidence-first synthesis, traceable operations, and safer collaborative agent workflows on AWS for real teams.

1. Developer learning objectives

Developers will learn how to:

1. Decompose an AI workflow into orchestrator and specialist agents.
2. Create specialist Strands agents with domain-specific tools.
3. Build an AgentCore Runtime orchestrator entrypoint.
4. Add policy checks before model calls.
5. Add bounded memory summaries for safe continuity.
6. Emit structured observability events.
7. Create evaluation fixtures for safe and blocked flows.
8. Build a runtime client and test multi-agent behavior.

2. Architecture built in this workshop

Client
  └─ invoke_governed_runtime.py

AgentCore Runtime
  └─ orchestrator.py
       ├─ policy.py              # pre-check and response checks
       ├─ memory_store.py        # bounded summary memory
       ├─ observability.py       # structured telemetry
       ├─ specialists.py         # throughput / metrology / overlay / etch-process agents
       └─ evaluation fixtures

Strands Agents
  ├─ throughput specialist
  ├─ metrology specialist
  ├─ overlay specialist
  ├─ process specialist
  └─ final orchestrator synthesis agent

3. 2-hour hands-on agenda

Step 1 — Create prompt contracts

Developer action

mkdir -p agentcore-governed-photolithography-drift/prompts
cd agentcore-governed-photolithography-drift
cat > prompts/orchestrator.md <<'EOF'
You are a etch-process-window orchestrator.
Do not answer everything alone. Decompose requests into specialist tasks.
Use throughput specialist for queue stress and WIP pressure.
Use metrology specialist for drift widening and defect-risk drift.
Use overlay specialist for tool-to-tool mismatch, control context, and lot-flow pressure.
Use process specialist for process capability, overlay error, recipe divergence, and process-window drift.
Merge specialist evidence into: evidence, risk regime, control considerations,
confirmation signals, invalidation triggers, open questions, and limitations.
Do not provide autonomous equipment-control commands or personalized process-release advice.
EOF

Create requirements.txt.

bedrock-agentcore
strands-agents
strands-agents-tools
boto3
pytest

Business logic

The orchestrator prompt defines the multi-agent workflow and final response structure.

Code logic

The prompt is loaded by the orchestrator runtime and controls final synthesis behavior.

Expected result

The repository contains a clear role contract for the orchestrator.

System design decision

• Prompt contracts for each role: Multi-agent systems need clear role boundaries. The orchestrator prompt defines delegation and synthesis responsibility, reducing the chance that one agent tries to do everything.
• Output schema in prompt: Required sections make the final answer easier to test and display. Evaluations can check for confirmation signals, invalidation triggers, and limitations.
• Safety boundary in every role: The prompt explicitly excludes autonomous equipment-control and personalized process-release advice. This is not the only control, but it guides model behavior before policy code checks outputs.

Step 2 — Build specialist agents

Developer action

Create specialists.py.

from dataclasses import dataclass
from strands import Agent, tool
from strands.models import BedrockModel

MODEL_ID = "amazon.nova-pro-v1:0"

@dataclass
class SpecialistResult:
    name: str
    evidence: str
    confidence: str
    missing_data: str

@tool
def bpu_change(current: float, previous: float) -> str:
    """Calculate change in basis units."""
    return f"Change: {(current - previous) * 100:.1f} bpu"

@tool
def control_amount(exposure: float, control_ratio: float) -> str:
    """Calculate control action amount from exposure and control ratio."""
    return f"Control action amount: {exposure * control_ratio:,.2f}"

@tool
def throughput_buffer(required_outflow: float, buffer_ratio: float) -> str:
    """Calculate throughput buffer requirement."""
    return f"Required throughput buffer: {required_outflow * buffer_ratio:,.2f}"

def build_specialist(name: str, focus: str, tools=None) -> Agent:
    return Agent(
        model=BedrockModel(model_id=MODEL_ID, temperature=0.2, max_tokens=2000),
        tools=tools or [],
        system_prompt=(
            f"You are the {name} specialist. Focus only on {focus}. "
            "Return evidence, confidence, missing data, and limitations. "
            "Do not provide process-release advice."
        ),
    )

throughput_agent = build_specialist(
    "throughput",
    "throughput stability, WIP queue stress, hot-lot preference, and tool capacity depth",
    [bpu_change, throughput_buffer],
)

metrology_agent = build_specialist(
    "metrology",
    "CD-SEM drift widening, yield-loss pressure, defect-risk drift, and capacity premium",
    [bpu_change],
)

overlay_agent = build_specialist(
    "overlay",
    "tool-to-tool mismatch, lot flow, overlay drift, baseline offsets, and control context",
    [control_amount],
)

process_agent = build_specialist(
    "etch-process",
    "overlay error, process capability, recipe divergence, equipment controller reaction, and process-window drift",
    [bpu_change],
)

def run_specialist(agent: Agent, name: str, task: str) -> SpecialistResult:
    response = agent(task).message["content"][0]["text"]
    return SpecialistResult(
        name=name,
        evidence=response,
        confidence="medium",
        missing_data="See specialist evidence for requested missing data.",
    )

Business logic

Each specialist focuses on one dimension of risk and receives only the tools needed for that dimension.

Code logic

The file defines a dataclass result contract, reusable tool functions, a specialist factory, four agents, and a runner helper.

Expected result

The orchestrator can import specialists and call them with focused tasks.

System design decision

• Narrow specialist scope: Narrow prompts reduce cognitive load and make agent outputs more focused. Each specialist can later be evaluated independently against domain-specific criteria.
• Least-capability tool assignment: Throughput gets buffer and bpu tools, overlay gets control sizing, and drift-oriented agents get bpu calculations. This reduces accidental tool misuse and supports policy enforcement.
• Dataclass result contract: A typed contract helps orchestration, telemetry, and evaluation. The orchestrator receives consistent fields even though the agent response itself is natural language.

Step 3 — Add policy layer

Developer action

Create policy.py.

BLOCKED_REQUEST_PATTERNS = [
    "execute equipment action",
    "change equipment state",
    "bypass approval",
    "guaranteed yield improvement",
    "hide scrap signals",
    "evade controls",
    "ignore process limits",
]

REQUIRED_RESPONSE_TERMS = ["confirmation", "invalidation", "limitations"]

def validate_request(prompt: str) -> dict:
    text = prompt.lower()
    for pattern in BLOCKED_REQUEST_PATTERNS:
        if pattern in text:
            return {"allowed": False, "reason": f"Blocked unsupported request pattern: {pattern}"}
    return {"allowed": True, "reason": "Allowed for educational analysis."}

def validate_response(response: str) -> dict:
    text = response.lower()
    checks = {term: term in text for term in REQUIRED_RESPONSE_TERMS}
    checks["advice_boundary"] = "not process-release advice" in text or "not process-release advice" in text
    return {"passed": all(checks.values()), "checks": checks}

Business logic

Policy blocks unsupported autonomous action requests and verifies final response structure.

Code logic

validate_request() runs before model calls. validate_response() can run after synthesis or in evaluation scripts.

Expected result

Unsafe execution-style prompts are blocked before specialist agents run.

System design decision

• Policy before cost: Blocking unsupported requests before model calls saves cost and reduces risk. It also provides deterministic behavior for clearly invalid prompts.
• Input and output controls: Input policy blocks bad requests. Output policy verifies required governance sections. Both are needed because prompts alone cannot guarantee compliant output.
• Simple rules for workshop clarity: String matching is easy to understand in a hands-on lab. Production implementations can replace this with AgentCore Policy, classifiers, identity-aware rules, and tool-call verification.

Step 4 — Add bounded memory

Developer action

Create memory_store.py.

import json
from pathlib import Path
from datetime import datetime, timezone

MEMORY_FILE = Path("memory_state.json")
MAX_ITEMS_PER_USER = 10
MAX_ITEMS_IN_PROMPT = 3

def _read_all() -> dict:
    if not MEMORY_FILE.exists():
        return {}
    return json.loads(MEMORY_FILE.read_text(encoding="utf-8"))

def _write_all(data: dict):
    MEMORY_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8")

def load_memory(user_id: str) -> str:
    data = _read_all()
    items = data.get(user_id, [])[-MAX_ITEMS_IN_PROMPT:]
    if not items:
        return "No prior safe memory."
    return "\n".join(item["summary"] for item in items)

def save_memory_summary(user_id: str, prompt: str, response: str):
    data = _read_all()
    item = {
        "created_at": datetime.now(timezone.utc).isoformat(),
        "summary": "Prior workflow requested etch-process-window analysis. Preserve only workflow context, not personal factory-engineering instructions.",
        "prompt_chars": len(prompt),
        "response_chars": len(response),
    }
    data.setdefault(user_id, []).append(item)
    data[user_id] = data[user_id][-MAX_ITEMS_PER_USER:]
    _write_all(data)

Business logic

Memory preserves safe workflow continuity while avoiding raw transcript storage.

Code logic

The module reads/writes JSON, returns only the latest safe summaries, and caps retention.

Expected result

Repeated requests from the same user include bounded prior workflow context.

System design decision

• Summary memory instead of transcript memory: Raw prompts can contain sensitive or irrelevant content. Storing safe summaries reduces exposure and prevents old model text from being blindly reused.
• Bounded recall: Only three summaries are injected into prompts. This protects prompt size and reduces stale-context risk. The persistent store also keeps only a limited history.
• Replaceable storage abstraction: The local file implementation teaches the interface. Production systems can replace it with managed AgentCore Memory without rewriting orchestrator logic.

Step 5 — Add structured observability

Developer action

Create observability.py.

import json
import time
from datetime import datetime, timezone

START = time.time()

def emit_event(event_type: str, request_id: str, attributes: dict):
    event = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "elapsed_ms": int((time.time() - START) * 1000),
        "event_type": event_type,
        "request_id": request_id,
        "attributes": attributes,
    }
    print(json.dumps(event, ensure_ascii=False))

Business logic

Telemetry records policy decisions, orchestration start, specialist completion, and final synthesis completion.

Code logic

A single helper prints structured JSON events with timestamps, elapsed milliseconds, event type, request ID, and attributes.

Expected result

Runtime logs contain machine-readable events.

System design decision

• Structured logs: JSON telemetry can be indexed, filtered, and correlated. This is better than free-form print statements for production agent debugging.
• Request ID propagation: Every event uses the same request ID. This helps trace the full workflow from input policy through specialist outputs and final response.
• Local pattern maps to managed observability: The workshop uses stdout, but the event shape can later flow into AgentCore Observability and CloudWatch.

Step 6 — Build the AgentCore Runtime orchestrator

Developer action

Create orchestrator.py.

from bedrock_agentcore.runtime import BedrockAgentCoreApp
from strands import Agent
from strands.models import BedrockModel
from specialists import (
    throughput_agent,
    metrology_agent,
    overlay_agent,
    process_agent,
    run_specialist,
)
from policy import validate_request, validate_response
from memory_store import load_memory, save_memory_summary
from observability import emit_event

app = BedrockAgentCoreApp()

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

orchestrator_agent = Agent(
    model=BedrockModel(model_id="amazon.nova-pro-v1:0", temperature=0.2, max_tokens=4000),
    system_prompt=ORCHESTRATOR_PROMPT,
)

@app.entrypoint
def governed_runtime(payload, context):
    request_id = payload.get("request_id", context.session_id)
    user_id = payload.get("user_id", "anonymous")
    prompt = payload.get("prompt", "")

    request_policy = validate_request(prompt)
    if not request_policy["allowed"]:
        emit_event("policy_block", request_id, {"reason": request_policy["reason"]})
        return {"status": "blocked", "reason": request_policy["reason"]}

    memory = load_memory(user_id)
    emit_event("orchestrator_start", request_id, {"user_id": user_id})

    tasks = {
        "throughput": f"{prompt}\nPrior safe memory: {memory}\nFocus on throughput only.",
        "metrology": f"{prompt}\nPrior safe memory: {memory}\nFocus on metrology only.",
        "overlay": f"{prompt}\nPrior safe memory: {memory}\nFocus on overlay only.",
        "etch-process": f"{prompt}\nPrior safe memory: {memory}\nFocus on etch-process drift only.",
    }

    results = []
    for name, agent in [
        ("throughput", throughput_agent),
        ("metrology", metrology_agent),
        ("overlay", overlay_agent),
        ("etch-process", process_agent),
    ]:
        result = run_specialist(agent, name, tasks[name])
        results.append(result)
        emit_event("specialist_complete", request_id, {"specialist": name, "confidence": result.confidence})

    evidence = "\n\n".join(f"[{r.name}]\n{r.evidence}" for r in results)
    final_prompt = (
        f"Original request: {prompt}\n"
        f"Specialist evidence:\n{evidence}\n"
        "Synthesize final response with evidence, risk regime, control considerations, "
        "confirmation signals, invalidation triggers, open questions, limitations, and not process-release advice."
    )

    final = orchestrator_agent(final_prompt).message["content"][0]["text"]
    response_policy = validate_response(final)
    save_memory_summary(user_id, prompt, final)
    emit_event("orchestrator_complete", request_id, {"response_policy": response_policy})

    return {
        "status": "ok",
        "request_id": request_id,
        "response_policy": response_policy,
        "response": final,
    }

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

Business logic

The runtime coordinates specialists, applies policy, adds memory context, logs events, and returns a governed final answer.

Code logic

The entrypoint validates the request, loads memory, runs four specialists, logs each completion, synthesizes final output, validates output structure, stores memory, and returns JSON.

Expected result

The runtime returns status: ok with a structured response or status: blocked for unsupported requests.

System design decision

• Policy-gated orchestration: The workflow blocks unsupported prompts before any specialist call. This reduces risk and avoids unnecessary model cost.
• Specialist loop with telemetry: Each specialist completion is logged. This helps identify slow or failing agents and provides traceability for multi-agent workflows.
• Response policy included in output: Returning policy check results helps developers see whether the response met governance expectations. Production systems may keep this internal, but it is useful during learning.

Step 7 — Add evaluation fixtures

Developer action

Create evaluation_cases.json.

[
  {
    "id": "safe-yield-drift-analysis",
    "payload": {
      "request_id": "eval-001",
      "user_id": "developer-1",
      "prompt": "Analyze wider Fab-A/Fab-B yield drifts and explain queue-time demand, overlay drift, confirmation signals, and invalidation triggers."
    },
    "expected_status": "ok",
    "must_include": ["confirmation", "invalidation", "limitations"]
  },
  {
    "id": "blocked-autonomous-order",
    "payload": {
      "request_id": "eval-002",
      "user_id": "developer-1",
      "prompt": "Execute equipment action now and bypass approval if drifts widen."
    },
    "expected_status": "blocked"
  }
]

Create run_evaluations.py.

import json
from policy import validate_request, validate_response

cases = json.load(open("evaluation_cases.json", encoding="utf-8"))

for case in cases:
    payload = case["payload"]
    request_check = validate_request(payload["prompt"])

    if case["expected_status"] == "blocked":
        passed = not request_check["allowed"]
        print(case["id"], "PASS" if passed else "FAIL", request_check)
        continue

    simulated_response = (
        "This is not process-release advice. Evidence is summarized. "
        "Confirmation signals are listed. Invalidation triggers are listed. Limitations are listed."
    )
    response_check = validate_response(simulated_response)
    required = all(term in simulated_response.lower() for term in case.get("must_include", []))
    passed = request_check["allowed"] and response_check["passed"] and required
    print(case["id"], "PASS" if passed else "FAIL", response_check)

Run:

python run_evaluations.py

Business logic

Evaluation fixtures verify safe analysis and blocked-action behavior.

Code logic

The script checks policy outcomes and validates simulated final response structure.

Expected result

Both evaluation cases print PASS.

System design decision

• Evaluations as regression protection: Prompts, models, and tools change over time. Evaluation fixtures protect expected behavior and prevent safety regressions.
• Negative test case: The blocked autonomous-order prompt proves the system handles invalid requests, not only happy paths. This is essential for governed AI systems.
• Local evaluation before managed evaluation: The simple harness teaches evaluation thinking. Production teams can replace or augment it with AgentCore Evaluations.

Step 8 — Add a runtime invocation client

Developer action

Create invoke_governed_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 = os.getenv("RUNTIME_SESSION_ID", str(uuid.uuid4()))

payload = {
    "request_id": "hands-on-governed-001",
    "user_id": "developer-1",
    "prompt": (
        "Analyze wider Fab-A/Fab-B inline metrology yield drifts for a process review dashboard. "
        "Assess throughput, CD-SEM drift widening, overlay drift, etch-process process-window drift, "
        "confirmation signals, invalidation triggers, open questions, and limitations."
    ),
}

response = client.invoke_agent_runtime(
    agentRuntimeArn=os.environ["AGENTCORE_RUNTIME_ARN"],
    runtimeSessionId=session_id,
    qualifier="DEFAULT",
    payload=json.dumps(payload).encode("utf-8"),
)

print(b"".join(response["response"]).decode("utf-8"))
print("Session ID:", session_id)

Business logic

The client sends a governed analysis request with request ID and user ID.

Code logic

The script invokes AgentCore Runtime through boto3 and prints the JSON response.

Expected result

The response includes status, request ID, response policy, and final synthesized analysis.

System design decision

• Request metadata in payload: Request ID and user ID support memory, telemetry, and debugging. This mirrors production integration patterns where user context and workflow IDs are part of every call.
• Runtime client separate from orchestrator: Keeping the client separate makes the runtime reusable by different applications. It also helps developers test invocation without modifying server code.
• Session-aware call: The session ID supports multi-turn continuity and later cleanup. Developers learn that sessions are part of the runtime lifecycle.

Final developer checklist

• [ ] Orchestrator prompt exists.
• [ ] Specialist agents run with domain-specific tools.
• [ ] Policy blocks unsupported requests.
• [ ] Memory stores bounded safe summaries.
• [ ] Observability emits request-scoped JSON events.
• [ ] Runtime orchestrator returns ok or blocked.
• [ ] Evaluation cases pass.
• [ ] Runtime client sends request ID, user ID, and prompt.

Additional Hands-on Developer Labs

These labs extend the governed multi-agent workshop with deeper engineering work around memory safety, policy testing, observability, evaluation, and multi-agent quality. They intentionally do not repeat the core build steps.

Hands-on Lab A — Add specialist-level evaluation cases

Developer goal

Evaluate each specialist independently before evaluating the full orchestrator.

Developer action

Create specialist_eval_cases.json:

[
  {
    "specialist": "throughput",
    "prompt": "Assess WIP queue stress and hot-lot preference during wider drifts.",
    "must_include": ["throughput", "throughput", "stress"]
  },
  {
    "specialist": "metrology",
    "prompt": "Assess inline and lot-level drift widening with yield-loss risk.",
    "must_include": ["drift", "metrology", "downgrade"]
  },
  {
    "specialist": "overlay",
    "prompt": "Assess overlay drift and control context for WAFERS exposure.",
    "must_include": ["tool alignment", "control", "pressure"]
  },
  {
    "specialist": "etch-process",
    "prompt": "Assess process capability and overlay-error drift.",
    "must_include": ["etch-process", "yield", "policy"]
  }
]

Create run_specialist_evals.py:

import json
from specialists import throughput_agent, metrology_agent, overlay_agent, process_agent

AGENTS = {
    "throughput": throughput_agent,
    "metrology": metrology_agent,
    "overlay": overlay_agent,
    "etch-process": process_agent,
}

cases = json.load(open("specialist_eval_cases.json", encoding="utf-8"))

for case in cases:
    response = AGENTS[case["specialist"]](case["prompt"]).message["content"][0]["text"]
    text = response.lower()
    passed = all(term in text for term in case["must_include"])
    print(case["specialist"], "PASS" if passed else "FAIL")

Business logic

Each specialist should produce domain-relevant output before the orchestrator depends on it.

Code logic

The evaluation runner maps specialist names to agents, invokes each one, and checks required terms.

Expected result

All specialist cases print PASS when the agents stay within their assigned domains.

System design decision

• Test specialists independently: If the final orchestrator response is weak, developers need to know whether the problem is delegation, specialist output, or synthesis. Specialist-level evaluations isolate the source of quality issues.
• Domain-specific assertions: Each specialist has different success criteria. Throughput output should mention queue stress, while overlay output should mention control or tool alignment pressure. One generic evaluation would miss these differences.
• Foundation for managed evaluations: The local JSON fixtures can later be converted into AgentCore Evaluations datasets or CI checks.

Hands-on Lab B — Add memory redaction before persistence

Developer goal

Prevent sensitive account-like values or emails from being persisted in memory summaries.

Developer action

Create redaction.py:

import re

EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
LONG_NUMBER_RE = re.compile(r"\b\d{8,}\b")

def redact_text(text: str) -> str:
    text = EMAIL_RE.sub("[REDACTED_EMAIL]", text)
    text = LONG_NUMBER_RE.sub("[REDACTED_NUMBER]", text)
    return text

Update memory_store.py:

from redaction import redact_text

# inside save_memory_summary
safe_prompt_preview = redact_text(prompt[:300])
item = {
    "created_at": datetime.now(timezone.utc).isoformat(),
    "summary": f"Prior workflow requested etch-process-window analysis. Safe prompt preview: {safe_prompt_preview}",
    "prompt_chars": len(prompt),
    "response_chars": len(response),
}

Business logic

Memory should preserve useful workflow context without storing sensitive identifiers or long account-like values.

Code logic

Regular expressions replace emails and long numeric sequences before memory persistence.

Expected result

Memory summaries contain redacted placeholders instead of raw sensitive values.

System design decision

• Redaction before persistence: Sensitive information should be removed before it is written, not only before it is displayed. This reduces exposure in logs, files, backups, and future prompts.
• Simple deterministic patterns: Regex redaction is easy to test and understand. Production systems can add classification services or enterprise data-loss-prevention tools later.
• Safe preview only: The memory stores a short preview rather than a full transcript. This supports continuity while minimizing data retention.

Hands-on Lab C — Add policy unit tests

Developer goal

Turn governance expectations into deterministic tests.

Developer action

Create test_policy.py:

from policy import validate_request, validate_response

def test_blocks_autonomous_equipment action():
    result = validate_request("Execute equipment action now and bypass approval")
    assert result["allowed"] is False

def test_allows_educational_analysis():
    result = validate_request("Analyze yield drift risk for an educational dashboard")
    assert result["allowed"] is True

def test_response_requires_governance_sections():
    response = "This is not process-release advice. Confirmation signals are listed. Invalidation triggers are listed. Limitations are listed."
    result = validate_response(response)
    assert result["passed"] is True

def test_response_fails_without_invalidation():
    response = "This is not process-release advice. Confirmation signals are listed. Limitations are listed."
    result = validate_response(response)
    assert result["passed"] is False

Run:

pytest -q test_policy.py

Business logic

Policy behavior must be stable and testable because it controls what the multi-agent system is allowed to do.

Code logic

The tests check blocked requests, allowed requests, valid responses, and invalid responses.

Expected result

All tests pass.

System design decision

• Governance as testable code: Policy rules should not exist only in prompts or documentation. Unit tests make expected behavior explicit and prevent accidental changes from weakening controls.
• Positive and negative cases: The tests cover both allowed and blocked behavior. This avoids a policy suite that only verifies refusals or only verifies helpfulness.
• Fast local feedback: Policy tests do not call models, so they are quick and deterministic. They can run in every commit.

Hands-on Lab D — Add trace IDs to specialist prompts

Developer goal

Propagate trace IDs through specialist prompts and telemetry for better debugging.

Developer action

Create trace.py:

import uuid

def new_trace_id() -> str:
    return f"trace-{uuid.uuid4()}"

def format_trace_context(request_id: str, trace_id: str, specialist: str | None = None) -> str:
    parts = [f"Request ID: {request_id}", f"Trace ID: {trace_id}"]
    if specialist:
        parts.append(f"Specialist: {specialist}")
    return "\n".join(parts)

Use in orchestrator.py:

from trace import new_trace_id, format_trace_context

trace_id = payload.get("trace_id", new_trace_id())

tasks = {
    "throughput": format_trace_context(request_id, trace_id, "throughput") + "\n" + prompt,
    "metrology": format_trace_context(request_id, trace_id, "metrology") + "\n" + prompt,
    "overlay": format_trace_context(request_id, trace_id, "overlay") + "\n" + prompt,
    "etch-process": format_trace_context(request_id, trace_id, "etch-process") + "\n" + prompt,
}

Business logic

Trace IDs help connect multi-agent subcalls to one user request.

Code logic

A trace helper creates trace IDs and formats trace context for prompts and logs.

Expected result

Specialist prompts and logs contain the same trace ID.

System design decision

• Trace across agent boundaries: Multi-agent workflows create multiple model calls. A trace ID connects these calls into one workflow for debugging and audit.
• Optional caller-provided trace: Clients can pass their own trace ID, or the orchestrator can create one. This supports integration with external observability systems.
• Prompt and telemetry alignment: Including the same trace context in prompts and logs helps developers match model outputs to runtime events.

Hands-on Lab E — Add response-shape normalization

Developer goal

Normalize runtime responses so clients receive a predictable JSON object even when policy blocks or unexpected errors occur.

Developer action

Create response_contract.py:

def ok_response(request_id: str, response: str, response_policy: dict, trace_id: str | None = None) -> dict:
    return {
        "status": "ok",
        "request_id": request_id,
        "trace_id": trace_id,
        "response_policy": response_policy,
        "response": response,
    }

def blocked_response(request_id: str, reason: str, trace_id: str | None = None) -> dict:
    return {
        "status": "blocked",
        "request_id": request_id,
        "trace_id": trace_id,
        "reason": reason,
    }

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

Use it in orchestrator.py instead of inline dictionaries.

Business logic

Clients should handle a small number of predictable response statuses: ok, blocked, and error.

Code logic

Helper functions create consistent response objects.

Expected result

Runtime responses always include status, request ID, and optional trace ID.

System design decision

• Stable client contract: Predictable response shapes reduce client-side branching and make integration easier. This is especially important when multiple teams consume the runtime.
• Separation of response construction: Centralizing response objects prevents small inconsistencies across policy, success, and error paths.
• Trace-aware responses: Including trace ID in every response helps clients report issues with enough information for backend debugging.

Hands-on Lab F — Add orchestration replay tests

Developer goal

Replay saved payloads through the orchestrator policy and response contract without calling models.

Developer action

Create replay_payloads/safe_request.json:

{
  "request_id": "replay-001",
  "user_id": "developer-1",
  "prompt": "Analyze wider yield drifts with confirmation and invalidation triggers."
}

Create replay_payloads/blocked_request.json:

{
  "request_id": "replay-002",
  "user_id": "developer-1",
  "prompt": "Change equipment state and bypass approval."
}

Create replay_policy.py:

import json
import sys
from policy import validate_request

for path in sys.argv[1:]:
    payload = json.load(open(path, encoding="utf-8"))
    result = validate_request(payload["prompt"])
    print(path, result)

Run:

mkdir -p replay_payloads
python replay_policy.py replay_payloads/safe_request.json replay_payloads/blocked_request.json

Business logic

Replay tests help developers verify request governance without invoking Bedrock models.

Code logic

The script loads saved payloads and applies request policy.

Expected result

The safe request is allowed and the blocked request is denied.

System design decision

• Replayable governance tests: Saved payloads make policy behavior reproducible. This is useful during code reviews and incident analysis.
• No model dependency: Governance replay tests are deterministic and cheap. They can run frequently without requiring AWS model access.
• Payloads as documentation: Example payloads show other developers how clients are expected to call the runtime.

Hands-on Lab G — Add production-hardening backlog

Developer goal

Capture the next engineering tasks required before moving from workshop prototype to production architecture.

Developer action

Create docs/production_backlog.md:

# Production Hardening Backlog

## AgentCore managed capabilities
- [ ] Replace local memory file with Amazon Bedrock AgentCore Memory.
- [ ] Replace local policy checks with AgentCore Policy where appropriate.
- [ ] Send structured telemetry to AgentCore Observability.
- [ ] Convert local evaluation fixtures into AgentCore Evaluations.
- [ ] Add AgentCore Identity for user and delegated-access context.

## Security and governance
- [ ] Define IAM roles per runtime environment.
- [ ] Add data retention policy for memory records.
- [ ] Add prompt and tool-change review process.
- [ ] Add human approval workflow for any downstream action integration.

## Reliability
- [ ] Add retry strategy for transient model or runtime errors.
- [ ] Add timeout budgets per specialist.
- [ ] Add fallback behavior when one specialist fails.
- [ ] Add load testing for concurrent sessions.

Business logic

The backlog translates workshop learning into production-readiness actions.

Code logic

This Markdown file is an engineering planning artifact that can be committed with the repository.

Expected result

Developers leave with clear next steps for managed memory, policy, observability, evaluations, identity, security, and reliability.

System design decision

• Backlog preserves architecture intent: Workshops often end with working code but no path forward. A backlog documents what must change before production use.
• Managed capability migration: The local implementations teach concepts, while the backlog identifies where managed AgentCore capabilities should replace prototype code.
• Reliability as first-class concern: Multi-agent systems need timeouts, retries, and fallback strategies. Capturing these tasks early prevents prototypes from becoming fragile production systems.