Agentic AI AcademyAgentic AI Academy
AgentCore Runtime/Lesson 3 of 6

Your First Agent with BedrockAgentCoreApp

Four lines to a deployable HTTP service

Beginner 14 minBuilder
What you'll be able to do
  • Install `bedrock-agentcore` and write the four-line agent shape: import, `app = BedrockAgentCoreApp()`, `@app.entrypoint`, `app.run()`
  • Explain what the SDK wraps — a thin Starlette/Uvicorn ASGI server that wires `POST /invocations` and `GET /ping`, session headers, streaming, and async task tracking
  • Read the request: a `payload` dict (e.g. `payload.get('prompt')`) plus the optional `RequestContext` second argument exposing `session_id` and `request_headers`
  • Return JSON (e.g. `{'result': ...}`) and understand how `app.run()` binds `0.0.0.0:8080` inside an ARM64 container
  • Test the exact cloud HTTP contract locally with `curl -X POST http://localhost:8080/invocations`
  • Describe the `GET /ping` health contract — HTTP 200 with `{status, time_of_last_update}` — and why `time_of_last_update` matters
At a glance

This is the hands-on core of AgentCore Runtime: turning a plain Python function into a deployable HTTP service in four lines. You'll import `BedrockAgentCoreApp`, decorate an entrypoint, return JSON, and call `app.run()` — and learn exactly what the SDK wires up underneath (the `/invocations` + `/ping` contract, session headers, streaming, async tracking). The payoff is that you can run and `curl` the *exact* cloud contract on `localhost:8080` before you deploy a single byte to AWS.

  1. 1Four lines to a deployable HTTP service
  2. 2What the SDK actually wraps
  3. 3Reading the request: payload and RequestContext
  4. 4Streaming responses and the /ping contract
  5. 5app.run() and testing the exact cloud contract locally
  6. 6From localhost to the cloud invoke API

Four lines to a deployable HTTP service

AgentCore Runtime hosts a container that speaks a known HTTP contract — not a framework. So the only thing the Python SDK has to do is turn your function into that contract. The package is bedrock-agentcore, and the wrapper is BedrockAgentCoreApp.

Here is a complete, runnable agent. Count the load-bearing lines:

python
from bedrock_agentcore.runtime import BedrockAgentCoreApp   # 1. import
from strands import Agent

app = BedrockAgentCoreApp()                                 # 2. instantiate
agent = Agent()

@app.entrypoint                                             # 3. decorate one function
def invoke(payload):
    result = agent(payload.get("prompt", "Hello"))
    return {"result": result.message}

if __name__ == "__main__":
    app.run()                                               # 4. serve

That's the "four lines to deploy" framing: (1) import BedrockAgentCoreApp, (2) app = BedrockAgentCoreApp(), (3) decorate a function with @app.entrypoint, (4) app.run(). Everything between the decorator and the return is your code — Strands here, but it could be LangGraph, CrewAI, or no framework at all. The decorator is the seam between AWS's managed HTTP server and your agent object.

Install it the usual way (Python ≥ 3.10):

bash
pip install bedrock-agentcore

The import path is from bedrock_agentcore.runtime import BedrockAgentCoreApp. You'll see some examples use the shorter from bedrock_agentcore import BedrockAgentCoreApp — both resolve to the same class; prefer the explicit .runtime path.

Key insight

The entrypoint is the only thing that changes between frameworks

Because Runtime hosts a protocol-speaking container, the invariant across every framework is identical: @app.entrypoint, read payload.get("prompt"), return {"result": ...}, app.run(). Swapping Strands for LangGraph changes the body of the function — graph.invoke({"messages": [...]}) instead of agent(prompt) — and nothing else. Learn this shape once and you've learned how to deploy any agent.

Note

Package versions move — pin and verify

[VOLATILE] As of writing the SDK bedrock-agentcore was reported around 1.13.0 and the separate bedrock-agentcore-starter-toolkit around 0.3.9 — verify the live versions on PyPI before pinning (https://pypi.org/project/bedrock-agentcore/). Note these are two different packages: this lesson uses the SDK (bedrock-agentcore); the toolkit is the deploy CLI covered later.

What the SDK actually wraps

BedrockAgentCoreApp is a thin Starlette/Uvicorn ASGI wrapper. It does not add an agent framework, an orchestration loop, or any opinion about your model. Its entire job is to translate your decorated function into the HTTP service contract that AgentCore Runtime expects, so AWS can route traffic to your container.

Concretely, the SDK wires up:

  • POST /invocations — the primary endpoint. Your @app.entrypoint function runs here. Input is JSON; output is JSON (or a stream — see below).
  • GET /ping — the health check the platform polls.
  • /ws (optional WebSocket) on the same port.
  • Session headers — it reads and echoes the HTTP session header X-Amzn-Bedrock-AgentCore-Runtime-Session-Id so related invocations land on the same microVM.
  • Streaming — an async generator entrypoint is automatically served as Server-Sent Events (text/event-stream).
  • Async task tracking@app.async_task / app.add_async_task(...) keep the session alive while background work runs.

The contract it implements is fixed: Host 0.0.0.0, Port 8080, Platform ARM64 for the HTTP protocol. You do not invent these — the SDK and the platform agree on them so your local server and the cloud runtime are byte-for-byte the same surface.

Because it's just an ASGI app, there's no magic to fear: app.run() starts Uvicorn, binds the port, and serves your two endpoints.

Tip

Different protocol, different port

The HTTP contract uses port 8080. If you instead host an MCP server in Runtime it's port 8000 (POST /mcp), A2A is 9000, and AG-UI is back on 8080. BedrockAgentCoreApp().run() defaults to the HTTP contract on 8080. Picking the wrong port for your declared protocol is a classic deploy failure.

Reading the request: payload and RequestContext

Your entrypoint receives the request body as a payload dict — the parsed JSON from the POST /invocations call. The convention across the AWS samples is a prompt key, read defensively:

python
@app.entrypoint
def invoke(payload):
    user_prompt = payload.get("prompt", "Hello")
    result = agent(user_prompt)
    return {"result": result.message}

The payload is your schema — prompt is a convention, not a requirement. You can send and read any JSON shape your agent expects.

The optional second argument: RequestContext. Add a second parameter and the SDK injects request metadata that isn't part of the body:

python
@app.entrypoint
def invoke(payload, context):
    session_id = context.session_id                       # the runtime session ID
    auth_header = context.request_headers.get("authorization")  # e.g. a bearer token
    result = agent(payload.get("prompt", "Hello"))
    return {"result": result.message, "session": session_id}

RequestContext exposes:

AttributeWhat it gives you
context.session_idThe caller's runtime session ID (used for microVM affinity and your own user-to-session bookkeeping).
context.request_headersThe inbound HTTP headers, e.g. authorization for OAuth-protected agents.

Returning a result. Return a JSON-serializable object — the SDK serializes it as the application/json response body. The samples use {"result": ...}; the raw HTTP contract documents shapes like {"response": "...", "status": "success"}. Either is fine as long as it's JSON your caller can parse.

Watch out

AWS does not map sessions to users — you do

context.session_id is the platform's microVM routing key, not an identity. AgentCore explicitly does not associate sessions with users; your client backend owns the user-to-session-ID relationship and any per-user session caps. Treat session_id as a routing/affinity token, and keep your own mapping if you need to know who is on a session.

Streaming responses and the /ping contract

Streaming is one decorator change. Make the entrypoint an async generator and the SDK automatically serves Content-Type: text/event-stream; every yield becomes one SSE data: event:

python
@app.entrypoint
async def handler(request: dict):
    agent = Agent()
    async for event in agent.stream_async(request.get("prompt", "")):
        yield event        # each yield -> one `data: {...}` SSE line

No manual SSE framing, no Content-Type juggling — the wrapper detects the generator and switches transport for you.

The GET /ping health contract. The platform polls /ping to decide whether your container is alive and whether to keep a session warm. The SDK answers it by default, but you can customize it:

python
from bedrock_agentcore.runtime import BedrockAgentCoreApp, PingStatus

app = BedrockAgentCoreApp()

@app.ping
def health():
    return PingStatus.HEALTHY        # or PingStatus.HEALTHY_BUSY

GET /ping must return HTTP 200 with a body like:

json
{ "status": "Healthy", "time_of_last_update": 1715000000 }

status is Healthy (ready) or HealthyBusy (busy with async work — this keeps the session alive past the idle timeout). time_of_last_update is a Unix-seconds timestamp and is REQUIRED: if you omit it, the idle timeout fires even while you're reporting HealthyBusy, and the platform kills the session mid-task.

Watch out

Never block the entrypoint

The platform's /ping poll runs on a separate thread. A long, blocking call inside @app.entrypoint starves that thread, so /ping stops responding and the session is killed around the 15-minute idle timeout — right in the middle of a long task. Offload long work to threads or async, and use @app.async_task / app.add_async_task(...) so /ping reports HealthyBusy while it runs.

app.run() and testing the exact cloud contract locally

app.run() starts the Uvicorn server. Locally it serves on http://localhost:8080; in a container it binds 0.0.0.0:8080 (the address AgentCore Runtime routes to). You can pass port= to change the local port if 8080 is taken.

The killer feature for development: the server you run locally implements the exact same HTTP contract as the deployed cloud runtime. So you test the real contract with plain curl — no AWS, no deploy, no auth:

bash
# Terminal 1: run the agent
python my_agent.py

# Terminal 2: hit the primary endpoint
curl -X POST http://localhost:8080/invocations \
  -H 'Content-Type: application/json' \
  -d '{"prompt":"What is the weather?"}'

# And the health endpoint the platform polls
curl http://localhost:8080/ping

A healthy /ping returns something like {"status": "Healthy", "time_of_last_update": 1715000000} with HTTP 200, and /invocations returns your JSON, e.g. {"result": "..."}.

One arch caveat for packaging. Runtime executes on AWS Graviton (ARM64). The deploy toolchains (CodeZip and the CLIs) handle the architecture for you, but if you hand-build a container yourself it MUST be linux/arm64:

bash
docker buildx build --platform linux/arm64 -t my-agent .

A wrong-arch (amd64) image is one of the most common deploy failures — it runs fine on your laptop and silently fails in the cloud. Local curl testing happens on your machine's native arch and is unaffected; the ARM64 requirement bites at deploy time.

Example

What the response body looks like over curl

For a non-streaming agent, POST /invocations returns application/json — your returned dict, e.g. {"result": "It's sunny."}. For an async-generator entrypoint, the same endpoint returns Content-Type: text/event-stream with lines like data: {"event": "..."} — so a streaming agent and a normal one differ only in how curl shows the body, not in the URL you hit.

Tip

Add OTEL early for free tracing

Add aws-opentelemetry-distro to your requirements.txt to get built-in tracing once deployed — wiring it from the start means your first cloud invocation already shows up in Observability with no extra code.

From localhost to the cloud invoke API

Once you've validated the contract locally, the cloud invocation is the same request through a different door. After deploy you get an agent runtime ARN, and you call the data-plane InvokeAgentRuntime API (boto3 client name bedrock-agentcore, method invoke_agent_runtime) with SigV4 auth:

python
import json, uuid, boto3

client = boto3.client("bedrock-agentcore")              # data plane
resp = client.invoke_agent_runtime(
    agentRuntimeArn="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/MyAgent-XXXX",
    runtimeSessionId=str(uuid.uuid4()),                 # >= 33 chars; reuse for context/affinity
    payload=json.dumps({"prompt": "Tell me a joke"}).encode(),
    qualifier="DEFAULT",                                 # endpoint alias
)
print(json.loads("".join(c.decode("utf-8") for c in resp.get("response", []))))

Notice the symmetry with the local curl: the body you sent as -d '{"prompt":"..."}' is exactly the payload here. The new parts are AWS-side concerns — the ARN, the runtimeSessionId (commonly cited as needing at least 33 characters — a UUID's 36 fits; confirm the current minimum in the InvokeAgentRuntime API reference), and the qualifier selecting an endpoint alias such as DEFAULT.

One gotcha to file away: if your agent uses OAuth inbound auth, you cannot use the AWS SDK to call InvokeAgentRuntime — you make a raw HTTPS request with the bearer token instead. SigV4-authenticated agents use the boto3 client above.

Key insight

Local and cloud are the same contract by design

This is why the four-line shape is so powerful: python my_agent.py + curl exercises the identical /invocations contract that invoke_agent_runtime hits in production. You debug payload shapes, streaming, and error handling on localhost in seconds, then deploy with confidence that the wire behavior won't change.

Try it: Build, run, and curl the four-line agent locally

Goal: write the minimal BedrockAgentCoreApp agent, run it, and prove you're hitting the exact cloud HTTP contract from your terminal — no AWS account required.

  1. Install the SDK. In a fresh virtualenv (Python >= 3.10): pip install bedrock-agentcore strands-agents. (Verify the version on https://pypi.org/project/bedrock-agentcore/.)
  2. Write the four lines. Create my_agent.py with from bedrock_agentcore.runtime import BedrockAgentCoreApp, app = BedrockAgentCoreApp(), an @app.entrypoint def invoke(payload): that reads payload.get("prompt", "Hello") and returns {"result": ...}, and if __name__ == "__main__": app.run(). (You can stub the agent — e.g. return {"result": f"echo: {payload.get('prompt')}"} — if you don't have Bedrock model access enabled yet.)
  3. Run it. python my_agent.py. Confirm it logs that it's serving on 0.0.0.0:8080 / localhost:8080.
  4. Curl the contract. From a second terminal:
    • curl http://localhost:8080/ping — confirm HTTP 200 and a body containing status and time_of_last_update.
    • curl -X POST http://localhost:8080/invocations -H 'Content-Type: application/json' -d '{"prompt":"What is the weather?"}' — confirm you get your {"result": ...} JSON back.
  5. Add RequestContext. Change the signature to def invoke(payload, context): and include context.session_id in the response. Re-run and re-curl, this time adding -H 'X-Amzn-Bedrock-AgentCore-Runtime-Session-Id: <a-36-char-uuid>'. Confirm that session ID appears in your response.
  6. Go streaming. Make a second entrypoint an async generator that yields a few chunks. Re-run, then curl -N -X POST http://localhost:8080/invocations -d '{"prompt":"count to 3"}' and observe data: SSE lines instead of a single JSON body.
  7. Reflect. In three sentences: what does BedrockAgentCoreApp add over a plain Starlette app, why does omitting time_of_last_update from /ping cause problems for long-running agents, and why must a hand-built container be linux/arm64?

Key takeaways

  1. 1The four-line shape: `from bedrock_agentcore.runtime import BedrockAgentCoreApp`, `app = BedrockAgentCoreApp()`, decorate a function with `@app.entrypoint`, and call `app.run()`. That's a deployable HTTP agent.
  2. 2`BedrockAgentCoreApp` is a thin Starlette/Uvicorn ASGI wrapper that wires the HTTP service contract: `POST /invocations`, `GET /ping`, optional `/ws`, session headers, streaming, and async task tracking. It's framework-agnostic.
  3. 3The entrypoint receives a `payload` dict (read `payload.get('prompt')`); add an optional second `RequestContext` argument for `context.session_id` and `context.request_headers` (e.g. the `authorization` header).
  4. 4Return JSON (the samples use `{'result': ...}`). `app.run()` serves `http://localhost:8080` locally and binds `0.0.0.0:8080` inside the container; the HTTP contract is Host `0.0.0.0`, port 8080, platform ARM64.
  5. 5Test the exact cloud contract locally: `curl -X POST http://localhost:8080/invocations -H 'Content-Type: application/json' -d '{"prompt":"..."}'`. The local server is byte-for-byte the deployed surface.
  6. 6`GET /ping` returns HTTP 200 with `{"status": "Healthy|HealthyBusy", "time_of_last_update": <unix_seconds>}`. The timestamp is required, and never block the entrypoint or you starve the `/ping` thread and get killed at the idle timeout.
  7. 7An `async`-generator entrypoint auto-serves SSE (`text/event-stream`), one event per `yield`; hand-built containers must be `linux/arm64`.

Quiz

Lock in what you learned

Check your understanding

0 / 4 answered

1.What does `BedrockAgentCoreApp` from the `bedrock-agentcore` package actually do?

2.Inside `@app.entrypoint def invoke(payload, context):`, where do you get the runtime session ID and the inbound `authorization` header?

3.Your agent does long background work and you report `HealthyBusy` from `/ping`, but the session still gets killed around 15 minutes. What's the most likely cause?

4.You want to test the EXACT contract your deployed agent will serve, before touching AWS. What do you run?

Go deeper

Hand-picked sources to keep learning