Streaming, Long-Running & Async Work

The minimal BedrockAgentCoreApp agent you've already built handles a synchronous request: a caller invokes /invocations, your @app.entrypoint runs, you return a dict, the caller gets one response. That's perfect for fast tool calls and short completions. But real agents do work that doesn't fit a single fast turn:

• A long completion you want to stream token-by-token so the user sees progress instead of staring at a spinner.
• A multi-step research or data job that takes minutes to hours — far longer than any caller wants to hold a connection open for.

AgentCore Runtime handles both with the same unified API — you don't deploy a different service for async work. What changes is the shape of your entrypoint and how you keep the session alive. Three shapes:

The Runtime decides how to serve your function by inspecting what the entrypoint is — a value-returning function is served as a normal response; an async generator is auto-served as SSE. You don't flip a config flag; you write the appropriate function and the SDK wires the contract.

To stream, make your entrypoint an async generator — a coroutine that yields. AgentCore Runtime detects this and automatically serves the response as text/event-stream (Server-Sent Events). The rule is simple and exact: each yield becomes one SSE data: event. You don't import an SSE library, set a Content-Type header, or format data: lines yourself — the SDK does it.

Most agent frameworks already expose an async streaming API; you just relay their events:

from bedrock_agentcore.runtime import BedrockAgentCoreApp
from strands import Agent

app = BedrockAgentCoreApp()

@app.entrypoint
async def handler(request: dict):
    agent = Agent()
    # Strands' stream_async yields incremental events; relay each one.
    async for event in agent.stream_async(request.get("prompt", "")):
        yield event   # one yield -> one SSE data: event

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

Because the response is SSE, the caller reads a stream of events rather than waiting for a single body. You can test the exact cloud contract locally — app.run() binds http://localhost:8080, and the same /invocations endpoint will now emit a stream:

curl -N -X POST http://localhost:8080/invocations \
  -H 'Content-Type: application/json' \
  -d '{"prompt":"Write a short poem about Graviton."}'

-N disables curl's buffering so you see events arrive incrementally. Each chunk on the wire is one data: event corresponding to one yield.

Streaming is the right tool when the total work still finishes inside the streaming window (max 60 min as of writing) and the value is showing progress. It is not the tool for a job that must outlive the connection — that's the long-running pattern below.

When a job takes minutes to hours, you cannot make the caller hold a connection open for it. The pattern AgentCore documents is:

1. Start the long task.
2. Respond immediately to the caller (e.g. an acknowledgement or a job handle).
3. Keep working in the background inside the same session.
4. Let the user poll for status/results on subsequent invocations of the same session.

The glue that makes step 3 possible is the /ping health endpoint and async task tracking. While the agent has background work in flight, /ping must report a busy status so Runtime knows the session is doing useful work and shouldn't be torn down at the idle timeout. Two SDK surfaces drive this:

• @app.ping — a handler that returns PingStatus.HEALTHY (ready/idle) or PingStatus.HEALTHY_BUSY (busy with async work; keeps the session alive past the idle timeout).
• @app.async_task / app.add_async_task("name") + app.complete_async_task(task_id) — async task tracking. While at least one tracked task is open, /ping automatically reports HealthyBusy.

Using the decorator, the SDK flips the status for you for the duration of the task:

import asyncio
from bedrock_agentcore.runtime import BedrockAgentCoreApp

app = BedrockAgentCoreApp()

@app.async_task
async def long_research(topic: str):
    # multi-step work that can run for minutes/hours.
    # While this task is open, /ping auto-reports HealthyBusy.
    await do_the_real_work(topic)

@app.entrypoint
async def handler(request: dict):
    action = request.get("action")
    if action == "start":
        asyncio.create_task(long_research(request["topic"]))   # don't await it
        return {"status": "started"}                           # respond immediately
    if action == "status":
        return {"status": check_progress()}                    # caller polls here

If you'd rather manage the lifecycle by hand instead of via the decorator, use the explicit pair:

task_id = app.add_async_task("long_research")   # /ping now reports HealthyBusy
try:
    await do_the_real_work(topic)
finally:
    app.complete_async_task(task_id)            # /ping returns to Healthy when none remain

Note that callers re-invoke the same session to poll. That's why reusing the runtimeSessionId across the start call and every poll call matters — it routes you back to the session that's holding your in-progress work. (A minimum session-id length of 33 characters is commonly cited; treat it as a guideline and confirm the exact constraint in the live API reference.)

Reporting HealthyBusy is necessary but not sufficient. The /ping response must return HTTP 200 with a body that includes time_of_last_update — and this field is Unix seconds and REQUIRED:

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

Here is the gotcha that bites people in production: if you omit time_of_last_update, the idle timeout fires even while you are reporting HealthyBusy. The status alone does not keep the session alive — Runtime uses the timestamp to know the busy state is current. A stale or missing timestamp is treated as no heartbeat, and the session is reaped mid-job.

So the discipline for a long-running agent is: advance time_of_last_update as your background work makes progress. A custom @app.ping handler returns both the status and a fresh timestamp. When you hand-roll @app.ping, you also own deciding busy-ness — track it yourself (e.g. a flag you set/clear around the work):

import time
from bedrock_agentcore.runtime import BedrockAgentCoreApp, PingStatus

app = BedrockAgentCoreApp()
_busy = False
_last_progress = time.time()

def mark_progress():
    global _last_progress
    _last_progress = time.time()   # call this as the background job advances

@app.ping
def ping():
    return {
        "status": PingStatus.HEALTHY_BUSY if _busy else PingStatus.HEALTHY,
        "time_of_last_update": int(_last_progress),   # Unix seconds, REQUIRED
    }

The simpler and safer path is to let the SDK manage /ping for you: when you use @app.async_task / add_async_task, the busy status is reported automatically while ≥1 task is open. If you hand-roll @app.ping, you own the whole contract — both the busy/idle status and always emitting a fresh time_of_last_update.

This is the single most important rule for long-running agents, and it's subtle because it has nothing to do with your business logic being correct.

The BedrockAgentCoreApp is a Starlette/Uvicorn ASGI app. The /invocations handler (your @app.entrypoint) and the /ping health endpoint share the same event loop / worker pool. A blocking call inside @app.entrypoint starves the /ping thread — Runtime stops getting healthy pings, decides the session is unresponsive, and kills it at 15 minutes during long work.

The trap is ordinary synchronous code that looks fine:

# BAD: blocks the event loop -> /ping is starved -> session reaped
@app.entrypoint
async def handler(request: dict):
    result = expensive_sync_call(request["data"])   # CPU-bound or blocking I/O
    return {"result": result}

Fix it by offloading the work so the event loop stays free to answer /ping:

import asyncio

@app.entrypoint
async def handler(request: dict):
    # Push blocking work onto a thread; the loop keeps serving /ping.
    result = await asyncio.to_thread(expensive_sync_call, request["data"])
    return {"result": result}

For genuinely long jobs, combine both ideas: offload AND track it as an async task so /ping reports HealthyBusy with a fresh timestamp the whole time:

@app.entrypoint
async def handler(request: dict):
    if request.get("action") == "start":
        # fire-and-forget background coroutine; respond immediately
        asyncio.create_task(run_tracked_job(request["data"]))
        return {"status": "started"}
    return {"status": check_progress()}

@app.async_task            # keeps /ping = HealthyBusy while open
async def run_tracked_job(data):
    # offload the blocking part so the loop never stalls
    await asyncio.to_thread(expensive_sync_call, data)

The mental model: the entrypoint's job is to hand off and return; the event loop's job is to keep answering /ping. Anything that holds the loop hostage is a latent session-killer.

Each shape has a hard time ceiling, and they're easy to confuse. As of writing ([VOLATILE] — verify on the live quotas page):

A few consequences worth internalizing:

• The 15-minute idle timeout and the 15-minute sync request timeout are different numbers that happen to match today. The idle timeout is what HealthyBusy + time_of_last_update defend against; the sync request timeout is the cap on a single request/response turn.
• The idle timeout is one of the few here you can adjust, via idleRuntimeSessionTimeout in LifecycleConfiguration. But raising it is not a substitute for correctly reporting HealthyBusy — it just changes how long an idle session survives.
• A job approaching the 8-hour async ceiling should checkpoint progress to durable storage (Memory, S3, a database), because session filesystem/memory is not durable and the session will end at the ceiling regardless.

Because these are volatile, design with headroom and read the value at deploy time rather than hardcoding assumptions. The long-running guide and the quotas page are the authoritative, live sources.