Your agent on AgentCore Runtime

@app.entrypoint
@requires_access_token(scopes=[OAUTH_SCOPE], provider_name=CREDENTIAL_PROVIDER_NAME, auth_flow="M2M")
async def invoke(payload: dict, context: RequestContext, access_token: str) -> dict | AsyncIterator[dict]:
    """`{"prompt": ...}` -> one JSON reply; with `"stream": true`, the events of `run_agent` as server-sent events.

    An optional `"variant"` picks a prompt variant (prompt_variants.py) for A/B comparison; default: the control.
    """
    actor_id, session_id = actor_from(context), context.session_id or "default_session"
    variant = variant_name(payload.get("variant"))
    events = run_agent(str(payload.get("prompt", "")), session_id, actor_id, access_token, variant)
    return events if payload.get("stream") else await last_event(events)


async def run_agent(
    prompt: str, session_id: str, actor_id: str, access_token: str, variant: str = ""
) -> AsyncIterator[dict]:
    """Stream `{"delta": text}` and `{"tool": name}` events as the orchestrator works, then the reply object."""
    specialist_results: list[str] = []
    result = None
    with gateway_client(access_token) as gateway:
        orchestrator = build_orchestrator(gateway, session_id, actor_id, specialist_results)
        variant = apply_variant(orchestrator, variant)
        async for event in orchestrator.stream_async(prompt):
            for item in stream_items(event):
                yield item
            result = event.get("result", result)
    answer = final_answer(str(result or ""), specialist_results)
    yield {"response": answer, "actor_id": actor_id, "session_id": session_id, "variant": variant}


def stream_items(event: dict) -> list[dict]:
    """What a streaming client sees of one Strands event: text deltas, and the tools the model decided to call."""
    if "data" in event:
        return [{"delta": event["data"]}]
    message = event.get("message") or {}
    if message.get("role") != "assistant":
        return []
    return [{"tool": block["toolUse"]["name"]} for block in message.get("content", []) if "toolUse" in block]


async def last_event(events: AsyncIterator[dict]) -> dict:
    """Drain a stream and keep only its final event (the reply object)."""
    reply: dict = {}
    async for event in events:
        reply = event
    return reply


def final_answer(text: str, specialist_results: list[str]) -> str:
    """The orchestrator's answer; if the model returned nothing, the last specialist answer (never an empty reply)."""
    if text.strip():
        return text
    return next((answer for answer in reversed(specialist_results) if answer.strip()), EMPTY_ANSWER)

LITELLM_BASE_URL = os.environ["LITELLM_BASE_URL"]


def make_model() -> OpenAIModel:
    """All model calls go through the LiteLLM gateway, keyless (the runtime role identifies you), guarded if enabled."""
    if guardrail.enabled():
        return guardrail.make_model(MODEL_ID, LITELLM_BASE_URL)
    return litellm_gateway.make_model(MODEL_ID, LITELLM_BASE_URL)

uv run bootcamp.py invoke "How many employees are in Engineering?" --actor alice-chen
uv run bootcamp.py invoke "And in Sales?" --actor alice-chen --session last   # same warm session: faster, keeps the conversation

uv run bootcamp.py invoke "Summarise our departments by headcount." --actor alice-chen --stream

Asking your agent as alice-chen (usually 20s-2m, up to ~5 min on a cold start)...
[calling data_agent]
| Department | Headcount |
|---|---:|
| Engineering | 481 |
| Sales | 360 |
...
(tools: data_agent; session: bootcamp-66772bb19e334da88e9450304bdbfaf5, actor: alice-chen; same session: --session last; spans: `traces`)

def stream_agent(outputs: dict[str, str], token: str, prompt: str, session_id: str) -> Iterator[dict]:
    """Ask with `"stream": true` and yield each server-sent event as it arrives (the last one is the reply object).

    An agent that does not stream yet answers with plain JSON, which is yielded as a single event.

    Raises:
        RuntimeError: The runtime answered with a non-2xx status.
    """
    payload, headers = {"prompt": prompt, "stream": True}, invoke_headers(token, session_id)
    url = runtime_url(outputs)
    with requests.post(url, json=payload, headers=headers, timeout=INVOKE_TIMEOUT, stream=True) as response:
        if not response.ok:
            raise RuntimeError(f"agent returned HTTP {response.status_code}: {response.text[:300]}")
        if "text/event-stream" not in response.headers.get("Content-Type", ""):
            yield response.json()
            return
        yield from sse_events(response.iter_lines(decode_unicode=True))


def sse_events(lines: Iterator[str]) -> Iterator[dict]:
    """The JSON objects of `data: ...` lines in a server-sent event stream (non-object data wrapped as "delta")."""
    for line in lines:
        if not line or not line.startswith("data:"):
            continue
        event = json.loads(line.removeprefix("data:").strip())
        yield event if isinstance(event, dict) else {"delta": str(event)}

uv run bootcamp.py invoke "Remember that I prefer reports as Python code." --actor alice-chen
# give extraction about a minute
uv run bootcamp.py invoke "How do I like my reports?" --actor alice-chen
uv run bootcamp.py invoke "How do I like my reports?" --actor jordan-lee

[PASS] stage 4 identity from the verified token only: HTTP {'none': 401, 'm2m': 401, 'forged-jwt': 403} | jordan-lee token + header alice-chen -> actor_id=jordan-lee
[PASS] stage 4 memory recall across sessions, no cross-user leak: bootcamp-check-0cb3dc07 recalled: Your favorite color is saffron. | jordan-lee forging bootcamp-check-0cb3dc07: leaked=False

uv run bootcamp.py deploy
uv run bootcamp.py invoke "How many people work in Sales?" --actor alice-chen
uv run bootcamp.py invoke "What is the capital of France?" --actor alice-chen
uv run bootcamp.py test --only 4

[CHALLENGE] SKIP stage 4 tools_used in the reply: the reply has no tools_used field yet
# after your change:
[CHALLENGE] PASS stage 4 tools_used in the reply: asked a database question: tools_used=['data_agent']