docs: Serverless Workers on AgentCore Runtime

docs/develop/python/workers/serverless-workers/agentcore.mdx

@@ -0,0 +1,280 @@
+---
+id: agentcore
+title: Serverless Workers on Amazon Bedrock AgentCore Runtime - Python SDK
+sidebar_label: Amazon Bedrock AgentCore
+description: Run a Temporal Worker on Amazon Bedrock AgentCore Runtime using the Python SDK.
+slug: /develop/python/workers/serverless-workers/agentcore
+toc_max_heading_level: 4
+tags:
+  - Workers
+  - Python SDK
+  - Serverless
+  - Amazon Bedrock AgentCore
+---
+
+import { ReleaseNoteHeader } from '@site/src/components'
+
+<ReleaseNoteHeader type="prerelease">
+  Amazon Bedrock AgentCore Runtime support is in Pre-release, and its APIs may change in backwards-incompatible ways.
+</ReleaseNoteHeader>
+
+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](/serverless-workers/agentcore).
+For the infrastructure procedure, see
+[Deploy a Serverless Worker on Amazon Bedrock AgentCore Runtime](/production-deployment/worker-deployments/serverless-workers/agentcore).
+
+## Install the AgentCore Runtime SDK {/* #install-agentcore-runtime-sdk */}
+
+Install the AgentCore Runtime SDK alongside the Temporal Python SDK:
+
+```bash
+pip install bedrock-agentcore
+```
+
+## Create a versioned Worker {/* #versioned-worker */}
+
+Serverless Workers require [Worker Versioning](/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:
+
+```python
+worker = Worker(
+    # ...
+    deployment_config=WorkerDeploymentConfig(
+        version=WorkerDeploymentVersion(
+            deployment_name=DEPLOYMENT_NAME,
+            build_id=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](/serverless-workers/agentcore#worker-versioning).
+
+Every Workflow needs a [versioning behavior](/worker-versioning#versioning-behaviors), 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 {/* #runtime-handler */}
+
+AgentCore Runtime invokes an HTTP handler. Use `BedrockAgentCoreApp` to provide that handler. Register the Worker as an
+asynchronous AgentCore task, then return an acknowledgment while the Worker continues polling in the background. The
+sample stores the background task in `_worker` and uses it to prevent another invocation from starting a duplicate
+Worker in the same Runtime session:
+
+<!--SNIPSTART python-agentcore-runtime-handler-->
+[bedrock_agentcore/strands_agent/agentcore_worker.py](https://github.com/temporalio/samples-python/blob/main/bedrock_agentcore/strands_agent/agentcore_worker.py)
+```py
+async def run_worker() -> None:
+    """Poll until idle, then drain."""
+    api_key = os.environ.get("TEMPORAL_API_KEY") or None
+    client = await Client.connect(
+        os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
+        namespace=os.environ.get("TEMPORAL_NAMESPACE", "default"),
+        api_key=api_key,
+        tls=bool(api_key),
+        plugins=[StrandsPlugin()],
+    )
+
+    tracker = ActivityTracker()
+    log.info("polling %s as %s/%s", TASK_QUEUE, DEPLOYMENT_NAME, BUILD_ID)
+    # execute_code is a sync Activity, so it needs an executor to block on.
+    with ThreadPoolExecutor(max_workers=4) as activity_executor:
+        worker = Worker(
+            client,
+            task_queue=TASK_QUEUE,
+            workflows=[workflows.StrandsAgentWorkflow],
+            activities=[execute_code],
+            activity_executor=activity_executor,
+            interceptors=[tracker],
+            deployment_config=WorkerDeploymentConfig(
+                version=WorkerDeploymentVersion(
+                    deployment_name=DEPLOYMENT_NAME, build_id=BUILD_ID
+                ),
+                use_worker_versioning=True,
+                default_versioning_behavior=VersioningBehavior.PINNED,
+            ),
+            graceful_shutdown_timeout=DRAIN,
+        )
+        async with worker:
+            await tracker.wait_until_idle(DEBOUNCE)
+    log.info("worker idle for %ss; drained", DEBOUNCE)
+
+
+async def _run_until_idle(task_id: int) -> None:
+    """Own the Worker's whole life, and always release the async task."""
+    try:
+        await run_worker()
+    except Exception:
+        # Nothing awaits this task, so an error would otherwise be swallowed.
+        log.exception("worker failed in async task")
+    finally:
+        # Without this the session stays HealthyBusy until MaxLifetime.
+        app.complete_async_task(task_id)
+
+
+@app.entrypoint
+async def invoke(payload: dict) -> dict:
+    """Start the Worker and acknowledge. The payload is unused."""
+    # Prevent duplicate workers since we exit early
+    global _worker
+    if _worker is not None and not _worker.done():
+        log.info("worker already polling %s", TASK_QUEUE)
+        return {"message": "worker already polling", "task_queue": TASK_QUEUE}
+
+    task_id = app.add_async_task("temporal-worker")
+    _worker = asyncio.create_task(_run_until_idle(task_id))
+
+    return {"message": "worker starting", "task_queue": TASK_QUEUE}
+
+
+```
+<!--SNIPEND-->
+
+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. `add_async_task` causes AgentCore to
+report the Runtime as busy while the Worker polls. `complete_async_task` releases that status after the Worker drains
+or fails.
+
+## Configure the Temporal connection {/* #configure-connection */}
+
+The `temporalio.envconfig` package loads [Temporal Client](/develop/python/client/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](/develop/environment-configuration).
+
+## Stop and drain the Worker {/* #stop-and-drain-the-worker */}
+
+AgentCore cannot tell when a Worker that is still polling has no Temporal work. The Runtime remains busy while the
+asynchronous task is registered, so it can remain active until its eight-hour maximum lifetime. To release capacity
+sooner, have the handler detect when the Worker has no useful work and complete the asynchronous task.
+
+When the condition remains true for an idle period, leave the `async with worker` block. The Worker stops polling for
+new Tasks and gives in-flight Activities time to complete before the Runtime handler returns.
+
+The following example from the
+[AgentCore sample Worker](https://github.com/temporalio/samples-python/blob/9d5c46bed0f0f6f8a726fa91f371c4c83f232ba2/bedrock_agentcore/strands_agent/agentcore_worker.py)
+defines an `ActivityTracker`. It uses an [Activity inbound Interceptor](/develop/python/workers/interceptors) to count
+running Activities.
+
+<!--SNIPSTART python-agentcore-activity-tracker-->
+[bedrock_agentcore/strands_agent/agentcore_worker.py](https://github.com/temporalio/samples-python/blob/main/bedrock_agentcore/strands_agent/agentcore_worker.py)
+```py
+# How long the Worker keeps polling after it goes idle.
+DEBOUNCE = float(os.environ.get("AGENTCORE_DEBOUNCE_SECONDS", "60"))
+# How long the drain waits for in-flight Activities (a model or tool call).
+DRAIN = timedelta(seconds=120)
+
+
+class ActivityTracker(Interceptor):
+    """Tracks in-flight activities and blocks until AGENTCORE_DEBOUNCE_SECONDS elapses with no events."""
+
+    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:
+        """Return once no Activity has run for ``debounce`` seconds."""
+        while True:
+            self.changed.clear()
+            try:
+                # Wake the moment an Activity starts or finishes; a timeout
+                # instead means nothing has happened for the whole window.
+                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
+    ) -> None:
+        super().__init__(next)
+        self._tracker = tracker
+
+    async def execute_activity(self, input: ExecuteActivityInput):
+        self._tracker.inflight += 1
+        self._tracker.changed.set()
+        log.info("activity in flight: %d", self._tracker.inflight)
+        try:
+            return await self.next.execute_activity(input)
+        finally:
+            self._tracker.inflight -= 1
+            self._tracker.changed.set()
+
+
+```
+<!--SNIPEND-->
+
+Register the tracker as a Worker Interceptor and wait for it inside the Worker context:
+
+```python
+tracker = ActivityTracker()
+worker = Worker(
+    client,
+    # ...
+    interceptors=[tracker],
+    graceful_shutdown_timeout=DRAIN,
+)
+
+async with worker:
+    await tracker.wait_until_idle(DEBOUNCE)
+```
+
+`ActivityTracker` retires the Worker only after 60 seconds without an Activity starting or completing and with no
+Activity running. A long-running Activity keeps the count above zero, so the idle policy does not interrupt it. The
+two-minute `graceful_shutdown_timeout` is a safety limit for any Activity still in flight when shutdown starts.
+
+Memory pressure can be another retirement condition. For example, the Runtime handler can monitor process memory and
+initiate the same graceful shutdown when usage crosses a threshold. Memory usage is not an idle signal. It tells you
+when to recycle a Worker, not whether it has work to do. Test any memory-based policy against the Runtime's memory
+limit and your Activity retry behavior.
+
+`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](/serverless-workers/agentcore#lifecycle).
+
+## Keep Activities safe across Worker termination {/* #activity-recovery */}
+
+AgentCore can end the compute that runs a Worker. An Activity running at that time can be interrupted and retried.
+Use [Activity Heartbeats](/develop/python/activities/timeouts#activity-heartbeats) so a retry resumes from its last
+recorded progress instead of starting over:
+
+```python
+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 {/* #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](/develop/python/platform/observability) and the
+[SDK metrics reference](/references/sdk-metrics).

docs/develop/python/workers/serverless-workers/index.mdx

@@ -14,10 +14,10 @@ tags:
 import { ReleaseNoteHeader } from '@site/src/components';
 
 <ReleaseNoteHeader type="publicPreview">
-  AWS Lambda support is in Public Preview. GCP Cloud Run support is in Pre-release, and its APIs may change in
-  backwards-incompatible ways. To request Cloud Run access, create a [support ticket](/evaluate/cloud/support#support-ticket) or
-  contact your account team, and [sign up for updates](https://temporal.io/pages/serverless-workers-updates) to hear
-  when Cloud Run reaches Public Preview.
+  AWS Lambda support is in Public Preview. Amazon Bedrock AgentCore Runtime and GCP Cloud Run support are in
+  Pre-release, and their APIs may change in backwards-incompatible ways. To request Cloud Run access, create a
+  [support ticket](/evaluate/cloud/support#support-ticket) or contact your account team, and
+  [sign up for updates](https://temporal.io/pages/serverless-workers-updates) to hear when Cloud Run reaches Public Preview.
 </ReleaseNoteHeader>
 
 Serverless Workers run on ephemeral, on-demand compute rather than long-lived processes.
@@ -29,4 +29,5 @@ For the end-to-end deployment guide, see [Deploy a Serverless Worker](/productio
 ## Supported providers
 
 - [**AWS Lambda**](/develop/python/workers/serverless-workers/aws-lambda) - Use the `lambda_worker` contrib package to run a Worker as a Lambda function. Covers setup, configuration, Lambda-tuned defaults, and observability.
+- [**Amazon Bedrock AgentCore Runtime**](/develop/python/workers/serverless-workers/agentcore) - Run a standard Worker from an AgentCore Runtime handler. Covers the handler, Worker Versioning, connection configuration, and Worker shutdown.
 - [**GCP Cloud Run**](/develop/python/workers/serverless-workers/cloud-run) - Run a standard Worker on a Cloud Run worker pool. Covers the versioned Worker setup, connection configuration, and handling scale-in.

docs/encyclopedia/workers/serverless-workers/aws-lambda.mdx

@@ -24,26 +24,38 @@ For a step-by-step deployment guide, see [Deploy a Serverless Worker on AWS Lamb
 
 ## Autoscaling {/* #autoscaling */}
 
-The Lambda autoscaling algorithm is event-driven and reactive.
+The autoscaling algorithm in this section applies to Serverless Workers on AWS Lambda and Amazon Bedrock AgentCore
+Runtime. The compute providers have different Worker lifecycles after a scale-out action. For AgentCore Runtime
+lifecycle details, see [Serverless Workers on Amazon Bedrock AgentCore Runtime](/serverless-workers/agentcore#lifecycle).
+
+The autoscaling algorithm is event-driven and reactive.
 Sync match failure is the primary control signal, and backlog aids sizing.
 
-When the [WCI](/serverless-workers#worker-controller-instance) needs more capacity, it calls the Lambda `InvokeFunction` API to start new Workers.
-Each call is a discrete action ("invoke N more functions"), not a target state.
-Temporal calls that API from outside your network, so no inbound connection to the function is needed.
-The WCI does not manage a fleet of instances.
+When the [WCI](/serverless-workers#worker-controller-instance) needs more capacity, it invokes the compute provider to
+start new Workers: the Lambda `InvokeFunction` API for Lambda or an AgentCore Runtime endpoint for AgentCore Runtime.
+Each call is a discrete action ("start N more Workers"), not a target state. Temporal calls the provider API from
+outside your network, so no inbound connection to the Worker is needed. The WCI does not manage a fleet of instances.
 
 ### Scale-out {/* #scale-out */}
 
-On sync match failure, the WCI invokes new Lambda functions.
-Because Lambda cold start is sub-second to low single-digit seconds, reactive-only control does not create meaningful backlog overshoot.
-The WCI can scale from zero with low latency.
+On sync match failure, the WCI starts new Workers through the compute provider API.
+
+For Lambda, cold start is sub-second to low single-digit seconds, so reactive-only control does not create meaningful
+backlog overshoot. The WCI can scale from zero with low latency.
+
+For AgentCore Runtime startup and session behavior, see
+[Lifecycle](/serverless-workers/agentcore#lifecycle).
 
 ### Scale-in {/* #scale-in */}
 
-Scale-in is automatic.
-Each Lambda invocation runs until the Worker has finished processing available Tasks or approaches the 15-minute execution time limit, then shuts down.
-There is no drain logic or stabilization window.
-The WCI does not need to actively remove capacity.
+The WCI does not maintain a target number of Workers or actively remove capacity. The provider and Worker lifecycle
+determine when a Worker stops.
+
+On Lambda, each invocation runs until the Worker has finished processing available Tasks or approaches the 15-minute
+execution time limit, then shuts down. There is no drain logic or stabilization window.
+
+On AgentCore Runtime, Worker shutdown and AgentCore session lifecycle settings determine when a Worker stops. See
+[Lifecycle](/serverless-workers/agentcore#lifecycle).
 
 ### Instance model {/* #instance-model */}