Skip to main content

Serverless Workers on Amazon Bedrock AgentCore Runtime - Python SDK

View Markdown

On Amazon Bedrock AgentCore Runtime, you run a standard long-lived Python Worker inside an AgentCore Runtime handler. Temporal starts the handler when the Worker Controller Instance needs capacity. The handler starts a Worker that polls the Task Queue, then stops it when your idle policy decides to release capacity.

The Worker uses the normal Python SDK. The handler uses the bedrock-agentcore package to receive AgentCore Runtime invocations.

For the provider behavior, including autoscaling, Worker Versioning, and the Runtime session lifecycle, see Serverless Workers on Amazon Bedrock AgentCore Runtime.

Install the AgentCore Runtime SDK​

Install the AgentCore Runtime SDK alongside the Temporal Python SDK:

pip install bedrock-agentcore

Create a versioned Worker​

Serverless Workers require Worker Versioning. Create the Worker as you would any long-lived Python Worker, then set deployment_config to declare its Worker Deployment Version and enable versioning:

import asyncio
import os
from datetime import timedelta

from temporalio.client import Client
from temporalio.common import VersioningBehavior, WorkerDeploymentVersion
from temporalio.worker import (
    ActivityInboundInterceptor,
    ExecuteActivityInput,
    Interceptor,
    Worker,
    WorkerDeploymentConfig,
)

from my_activities import my_activity
from my_workflows import MyWorkflow

DEBOUNCE = float(os.environ.get("AGENTCORE_DEBOUNCE_SECONDS", "60"))
DRAIN = timedelta(seconds=120)


class ActivityTracker(Interceptor):
    def __init__(self) -> None:
        self.inflight = 0
        self.changed = asyncio.Event()

    def intercept_activity(
        self, next: ActivityInboundInterceptor
    ) -> ActivityInboundInterceptor:
        return TrackedActivity(next, self)

    async def wait_until_idle(self, debounce: float) -> None:
        while True:
            self.changed.clear()
            try:
                await asyncio.wait_for(self.changed.wait(), timeout=debounce)
            except asyncio.TimeoutError:
                if self.inflight == 0:
                    return


class TrackedActivity(ActivityInboundInterceptor):
    def __init__(self, next: ActivityInboundInterceptor, tracker: ActivityTracker):
        super().__init__(next)
        self.tracker = tracker

    async def execute_activity(self, input: ExecuteActivityInput):
        self.tracker.inflight += 1
        self.tracker.changed.set()
        try:
            return await self.next.execute_activity(input)
        finally:
            self.tracker.inflight -= 1
            self.tracker.changed.set()


def create_worker(client: Client, tracker: ActivityTracker) -> Worker:
    return Worker(
        client,
        task_queue=os.environ["TEMPORAL_TASK_QUEUE"],
        workflows=[MyWorkflow],
        activities=[my_activity],
        interceptors=[tracker],
        graceful_shutdown_timeout=DRAIN,
        deployment_config=WorkerDeploymentConfig(
            version=WorkerDeploymentVersion(
                deployment_name=os.environ["TEMPORAL_DEPLOYMENT_NAME"],
                build_id=os.environ["TEMPORAL_BUILD_ID"],
            ),
            use_worker_versioning=True,
            default_versioning_behavior=VersioningBehavior.PINNED,
        ),
    )

TEMPORAL_DEPLOYMENT_NAME and TEMPORAL_BUILD_ID must match the Worker Deployment Version that you create with temporal worker deployment create-version. Configure that Worker Deployment Version with the AgentCore Runtime endpoint that Temporal invokes. For the endpoint configuration, see Worker Versioning.

Every Workflow needs a versioning behavior, either PINNED or AUTO_UPGRADE. Setting default_versioning_behavior as shown applies PINNED behavior to every Workflow on the Worker. To set the behavior per Workflow instead, pass versioning_behavior to the @workflow.defn decorator.

Start the Worker from the Runtime handler​

AgentCore Runtime invokes an HTTP handler. Use BedrockAgentCoreApp to provide that handler, and use async_task so AgentCore keeps the Runtime active while the Worker polls:

from bedrock_agentcore.runtime import BedrockAgentCoreApp
from temporalio.client import Client
from temporalio.envconfig import ClientConfig

app = BedrockAgentCoreApp()


@app.entrypoint
@app.async_task
async def invoke(_: dict) -> dict:
    client = await Client.connect(**ClientConfig.load_client_connect_config())
    tracker = ActivityTracker()
    worker = create_worker(client, tracker)

    async with worker:
        await tracker.wait_until_idle(DEBOUNCE)

    return {"message": "Worker drained"}

The payload does not represent a Workflow input. The Worker Controller Instance invokes the endpoint to add Worker capacity. Applications start Workflows through the Temporal Client, as usual.

Configure the Temporal connection​

The temporalio.envconfig package loads Temporal Client configuration from environment variables and an optional TOML configuration file. Set the Temporal address, Namespace, Task Queue, and Worker Deployment Version values as Runtime environment variables. Store a Temporal Cloud API key or TLS material in a secret store rather than in the Runtime definition.

For the supported connection variables, config-file format, and profiles, see Environment configuration.

Stop and drain the Worker​

Use ActivityTracker to keep a Worker available while it runs Activities and stop it after a period with no Activity work:

DEBOUNCE = float(os.environ.get("AGENTCORE_DEBOUNCE_SECONDS", "60"))
DRAIN = timedelta(seconds=120)

tracker = ActivityTracker()
worker = create_worker(client, tracker)

async with worker:
    await tracker.wait_until_idle(DEBOUNCE)

ActivityTracker starts a 60-second timer when no Activity is running. Starting or completing an Activity resets the timer. When the timer expires, the Worker leaves the async with block, stops polling, and waits up to two minutes for in-flight Activities to complete.

This policy is appropriate when Activities represent the work that should keep the Worker available. The AgentCore sample runs model and tool calls as Activities. It is not a universal definition of idleness. If your Worker has a different signal, implement an idle policy based on that signal.

AGENTCORE_DEBOUNCE_SECONDS controls the idle period. graceful_shutdown_timeout controls how long the Worker waits for in-flight Activities after it stops polling. Choose both values for your workload, and account for AgentCore's maximum Runtime lifetime. For the AgentCore lifecycle settings, see Lifecycle.

Keep Activities safe across Worker termination​

AgentCore can end the compute that runs a Worker. An Activity running at that time can be interrupted and retried. Use Activity Heartbeats so a retry resumes from its last recorded progress instead of starting over:

from temporalio import activity


@activity.defn
async def my_activity(items: list[str]) -> str:
    for i, item in enumerate(items):
        activity.heartbeat(i)
        # ... process item
    return "done"

Add observability​

An AgentCore Runtime Worker emits the same traces and metrics as a Worker on other compute. For metrics export and OpenTelemetry tracing interceptors, see Observability - Python SDK and the SDK metrics reference.