OpsfleetAgent Bootcamp

Task 4 · 8 tasks

Your agent on AgentCore Runtime

Deploy the Phase 1 orchestrator, wired to Gateway, Memory and LiteLLM. This is where you code.

35 minAdvanced
Alice’s ask

The “Agent Deployment and Tool Access” challenge

Everything is in place: identity, a tool server, a Gateway and memory. Now Alice's executive assistant itself goes live. It's the Phase 1 orchestrator with three swaps: tools come from the Gateway, sessions come from AgentCore Memory, and the human approval prompt becomes a read-only guard.

Platform

What the platform provisions for you

terminal
uv run bootcamp.py up 4
  • Your agent from phase2/app/agent/, deployed to AgentCore Runtime (HTTP protocol) with JWT inbound auth
  • Environment variables so the code has no hardcoded ids: GATEWAY_URL, MEMORY_ID, MEMORY_STRATEGY_ID, CREDENTIAL_PROVIDER_NAME, MODEL_ID and LITELLM_BASE_URL
  • An execution role awsworkshop-<name>-agent-runtime, which the LiteLLM gateway maps to your budget. No API key is deployed anywhere
You

What you do as a developer

  1. Read the entrypoint

    phase2/app/agent/agent.py
    @app.entrypoint
@requires_access_token(scopes=[OAUTH_SCOPE], provider_name=CREDENTIAL_PROVIDER_NAME, auth_flow="M2M")
def invoke(payload: dict, context: RequestContext, access_token: str) -> dict:
    actor_id = actor_from(context)
    gateway = MCPClient(
        lambda: streamable_http_client(
            GATEWAY_URL, http_client=create_mcp_http_client(headers={"Authorization": f"Bearer {access_token}"})
        )
    )
    sandbox_tools = optional_tools()
    with gateway:
        orchestrator = Agent(
            name="Executive Assistant",
            model=make_model(),
            system_prompt=orchestrator_prompt(actor_id, sandbox_tools),
            tools=[*build_specialists(gateway.list_tools_sync()), *sandbox_tools],
            session_manager=memory_session_manager(context.session_id or "default_session", actor_id),
            callback_handler=None,
        )
        result = orchestrator(payload.get("prompt", ""))
    return {"response": str(result)}

    @requires_access_token fetches a Gateway token through AgentCore Identity. Model calls use the same keyless gateway client as Phase 1:

    phase2/app/agent/agent.py
    LITELLM_BASE_URL = os.environ["LITELLM_BASE_URL"]


def make_model() -> OpenAIModel:
    """All model calls go through the LiteLLM gateway, keyless: the runtime's own IAM role identifies you."""
    return litellm_gateway.make_model(MODEL_ID, LITELLM_BASE_URL)

  2. Talk to your deployed agent

    terminal
    uv run bootcamp.py invoke "How many employees are in Engineering?" --actor alice-chen
uv run bootcamp.py invoke "What is the weather in Seattle?" --actor alice-chen

  3. Prove memory works across sessions and users

    Each invoke without --session starts a new session.

    terminal
    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

    Alice gets her preference back; Jordan doesn't.

  4. Personalise the assistant

    Challenge

    Change orchestrator_prompt so the assistant greets the user by name, answers in short bullet points, and politely declines requests unrelated to DataStream.

    Hint 1

    The prompt is built per request and already receives actor_id. Keep the routing rules; add style and scope rules.

    Hint 2

    Be explicit about the refusal (“If a request is not about DataStream, say so in one sentence”). Redeploy with uv run bootcamp.py deploy and test with an off-topic question.

    Solution
    phase2/app/agent/agent.py (starting point)
    def orchestrator_prompt(actor_id: str, sandbox_tools: list) -> str:
    """System prompt for the orchestrator, plus one hint per enabled optional sandbox tool."""
    return (
        f"You are CEO Alice's executive assistant at DataStream Corp, talking to {actor_id}.\n"
        "Route database questions to data_agent and weather questions to weather_agent; "
        "answer simple company questions directly. Use what you remember about the user.\n"
        f"{prompt_hints(sandbox_tools)}"
    ).strip()
    one possible solution
    def orchestrator_prompt(actor_id: str, sandbox_tools: list) -> str:
    return (
        f"You are CEO Alice's executive assistant at DataStream Corp, talking to {actor_id}. "
        "Greet them by name. Answer in at most five short bullet points.\n"
        "Route database questions to data_agent and weather questions to weather_agent; "
        "answer simple company questions directly. Use what you remember about the user.\n"
        "If a request is not about DataStream, decline politely in one sentence.\n"
        f"{prompt_hints(sandbox_tools)}"
    ).strip()

  5. Add a tool

    Challenge

    Alice asks “what's today's date?” and the agent guesses. Give the orchestrator a tool that returns the current date, and tell it when to use it.

    Hint 1

    Any function with @tool, type hints and a docstring is a tool. Add it to the orchestrator's tools list next to the specialists.

    Hint 2

    Use datetime.date.today().isoformat(). agent.py only imports os, so add import datetime at the top. Mention the tool in the prompt.

    Solution
    phase2/app/agent/agent.py
    import datetime


@tool
def today() -> str:
    """Return today's date (ISO 8601). Use it for any question about dates or deadlines."""
    return datetime.date.today().isoformat()


# in invoke():
tools=[*build_specialists(gateway.list_tools_sync()), *sandbox_tools, today],

  6. Experiments

    • Set MODEL_ID=claude-sonnet-5-5, deploy, and compare answers and cost (uv run bootcamp.py llm).
    • Ask for something destructive: invoke "Delete employee 5". The ReadOnlyGuardHook cancels the call and the agent explains why. Keep this in mind for Task 7.

  7. Deploy and re-test

    deploy rebuilds your code and re-applies the current stage with the MODEL_ID from .env. Then talk to it and re-run the stage checks:

    terminal
    uv run bootcamp.py deploy
uv run bootcamp.py invoke "What is today's date, and 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

Check your work

terminal
uv run bootcamp.py test --only 4
uv run bootcamp.py invoke "How many employees are in Engineering?" --actor alice-chen

Passes when the agent answers, a fact told in one session is recalled in another, and the model spend landed on your own LiteLLM budget.

Under the hood

The agent runs as an HTTP-protocol runtime: BedrockAgentCoreApp serves /invocations and the decorated function receives the JSON payload plus a RequestContext (session id, headers). The runtime sticks a session to one micro-VM, so consecutive calls in a session stay warm. Outbound, the agent's workload identity asks AgentCore Identity for an M2M token (credential provider), uses it to open an MCP session with the Gateway, and sends model calls to LiteLLM with a presigned STS identity from the runtime's execution role. The gateway maps awsworkshop-<name>-agent-runtime to your budgeted key, forwards to Bedrock and debits your budget.