Skip to content

Production Lifecycle Controls

A prompt against a single agent instance can, left unbounded, loop longer than you want, spend more tokens than you budgeted, or keep running after the caller has walked away. When you run agents in production, you need to cap that work, stop it on demand, and read back a clear reason for why it ended.

This page collects the controls the agent loop exposes for exactly that: bounding a single call, cancelling one mid-flight, interpreting the stop reason it returns, recovering from tool and model errors, and running one agent safely across many concurrent requests. Each control works on its own; the worked example at the end combines them into one request handler.

Cap the work one call can do

Section titled “Cap the work one call can do”

To bound a single invocation, pass a set of limits alongside the prompt. You can cap the number of turns (loop iterations), the output tokens, or the total tokens. All three are optional.

from strands import Agent


agent = Agent()


result = agent(
    "Summarize this document",
    limits={
        "turns": 5,
        "output_tokens": 2000,
        "total_tokens": 10000,
    },
)


if result.stop_reason == "limit_turns":
    print("Hit turn budget")
elif result.stop_reason == "limit_total_tokens":
    print("Hit token budget")

The same parameter works with invoke_async and stream_async. Each cap must be a positive integer.

Limits are checked at the top of each loop iteration, not mid-call. A single turn can overshoot its token budget, but the check fires before the next turn starts, and tools requested by the previous turn always finish first. When several caps trip at once, the reported stop reason follows priority order: turns, then total tokens, then output tokens.

Limits apply to the current invocation only. A reused agent starts each call with fresh counters, and the message history stays in a valid, reinvokable state, so you can retry with a higher budget.

Know why the loop stopped

Section titled “Know why the loop stopped”

Most invocations return a result with a stop reason that tells your application what to do next. Read it from result.stop_reasonresult.stopReason.

  • end_turnendTurn or stop_sequencestopSequence: return the completed result.
  • limit_turnslimitTurns, limit_total_tokenslimitTotalTokens, or limit_output_tokenslimitOutputTokens: report budget exhaustion or reinvoke with a higher limit.
  • cancelled: treat the result as a caller- or application-initiated stop. The incomplete assembled model message is omitted from the result and history. Streamed deltas already delivered to your consumer remain visible.
  • content_filteredcontentFiltered or guardrail_intervenedguardrailIntervened: apply your application’s policy rather than retrying blindly.
  • interrupt: collect the requested human input and resume the agent. See Interrupts.

The three limit reasons signal graceful budget exhaustion, not failure. Content filtering and guardrail intervention both terminate the loop and should be handled deliberately rather than retried blindly.

Stop a running agent

Section titled “Stop a running agent”

To stop a call from outside the loop, on a client disconnect, a timeout, or a “Stop” button, cancel it. The agent checks for cancellation at fixed checkpoints and returns a result with a cancelled stop reason. Use a per-invocation signal when the stop belongs to one request. Use agent.cancel()agent.cancel() when another component owns the running agent but not its signal.

import threading


from strands import Agent


agent = Agent()


cancel_signal = threading.Event()
deadline = threading.Timer(30.0, cancel_signal.set)
deadline.start()


try:
    result = agent(
        "Analyze this large dataset",
        cancel_signal=cancel_signal,
    )
finally:
    deadline.cancel()

Pass the same event to a request-disconnect or “Stop” button handler when cancellation is user-driven. The agent never sets or clears the caller-owned event. agent.cancel()agent.cancel() is thread-safe and idempotent, but only call it for a known-running invocation. Calling it while the agent is idle cancels the next invocation.

The agent checks for cancellation before tool execution and at safe loop boundaries. With sequential tool execution, it can skip tools that have not started. With the default concurrent executor, sibling tools may already be running. Once a tool starts, cancellation is cooperative: the tool decides whether to stop early. Long-running tools should poll the cancel signal between steps or forward it to an API that accepts one.

from strands import tool
from strands.types.tools import ToolContext




@tool(context=True)
def normalize_records(
    records: list[str],
    tool_context: ToolContext,
) -> str:
    """Normalize records while checking for cancellation between items."""
    normalized = []
    for record in records:
        if tool_context.cancel_signal.is_set():
            return f"cancelled after {len(normalized)} records"
        normalized.append(record.strip().upper())
    return "\n".join(normalized)

For the full checkpoint tables and provider-level abort behavior, see Cancellation in the agent loop reference. Cancellation stops the agent entirely; it differs from an interrupt, which pauses for human input and can resume.

Recover from tool and model errors

Section titled “Recover from tool and model errors”

A tool-body failure does not crash the loop by default. The execution system captures the error, returns it to the model as an error result, and lets the model adjust or try an alternative. Lifecycle hook failures, exhausted model retries, invalid configuration, and other SDK errors still propagate to your application. Treat returned stop reasons and raised exceptions as separate outcomes.

Two mechanisms give you more control:

  • Model provider errors such as rate limits retry automatically with exponential backoff. To change the attempt count, backoff, or which errors are retryable, configure a strategy. See Retry Strategies.
  • Tool-level recovery such as retrying a flaky tool, propagating an unexpected exception instead of feeding it back to the model, or capping how often a tool runs is done with Hooks. The hooks cookbook covers model call retry, tool call retry, and exception handling.

Handle a truncated model response

Section titled “Handle a truncated model response”

When a single model response exceeds the provider’s per-call output cap, the loop cannot continue from a partial message. This is distinct from the graceful limit_* stop reasons: the loop raises MaxTokensReachedExceptionMaxTokensError rather than returning a result.

The SDK adds the recovered partial message to conversation history before raising MaxTokensReachedException. Reduce the next request, raise the provider’s output cap, or split the task into smaller steps, then invoke the same agent again to continue from that history.

Run one agent across many requests

Section titled “Run one agent across many requests”

An agent mutates its conversation history as each invocation runs, so overlapping calls on one instance would interleave their messages and corrupt the history. By default the agent processes one invocation at a time and rejects overlap.

Invoking an agent that is already running raises ConcurrencyException. The concurrent_invocation_mode constructor parameter controls this: it takes a ConcurrentInvocationMode of either THROW (the default) or UNSAFE_REENTRANT.

from strands import Agent
from strands.types.exceptions import ConcurrencyException


agent = Agent()


try:
    result = agent("Summarize this report")
except ConcurrencyException:
    # Another invocation is already running on this agent instance
    ...

When a retry arrives for a request that is still in flight, deduplicate it rather than erroring or doing the work twice. Pass idempotency_token to any invocation method to identify the logical request. A second call with the same token waits for the original and returns the same AgentResult; a call with a different token still raises ConcurrencyException.

result = agent("Process order 1234", idempotency_token="order-1234")

UNSAFE_REENTRANT removes the single-invocation guard entirely, and nothing then protects the history from concurrent mutation. To run independent work in parallel, a separate agent per task is safer than sharing one.

Worked example: a high-concurrency service

Section titled “Worked example: a high-concurrency service”

A service that fields thousands of concurrent invocations puts every control above to work at once. The pattern that scales is one fresh agent per request: it sidesteps the concurrency guard entirely, since no two requests share an instance. Each request gets a bounded turn count, a per-run token budget, and a deadline, and the handler maps the stop reason onto a typed outcome the caller can branch on.

import threading
from dataclasses import dataclass
from enum import Enum


from strands import Agent
from strands.models import BedrockModel


MODEL_ID = "global.anthropic.claude-sonnet-5"




class Outcome(str, Enum):
    COMPLETED = "completed"
    BUDGET_EXCEEDED = "budget_exceeded"
    CANCELLED = "cancelled"
    BLOCKED = "blocked"
    STOPPED = "stopped"




@dataclass
class Reply:
    outcome: Outcome
    text: str | None




def classify(stop_reason: str) -> Outcome:
    if stop_reason in ("end_turn", "stop_sequence"):
        return Outcome.COMPLETED
    if stop_reason == "cancelled":
        return Outcome.CANCELLED
    if stop_reason in ("limit_turns", "limit_total_tokens", "limit_output_tokens"):
        return Outcome.BUDGET_EXCEEDED
    if stop_reason in ("content_filtered", "guardrail_intervened"):
        return Outcome.BLOCKED
    return Outcome.STOPPED




def handle_request(prompt: str, deadline_seconds: float = 20.0) -> Reply:
    """Run one request on its own agent with bounded turns, tokens, and time."""
    # A fresh agent per request keeps concurrent calls from sharing history
    agent = Agent(model=BedrockModel(model_id=MODEL_ID))


    cancel_signal = threading.Event()
    deadline = threading.Timer(deadline_seconds, cancel_signal.set)
    deadline.start()


    try:
        result = agent(
            prompt,
            limits={"turns": 6, "total_tokens": 20_000},
            cancel_signal=cancel_signal,
        )
    finally:
        deadline.cancel()


    outcome = classify(result.stop_reason)
    text = str(result) if outcome == Outcome.COMPLETED else None
    return Reply(outcome=outcome, text=text)

The handler does not raise when a budget or cancellation signal stops the agent. It returns a typed outcome and exposes response text only after normal completion. Errors that the SDK cannot recover from, such as a truncated model response, still propagate to the request boundary. This stateless pattern does not resume interrupts; applications that need human input must preserve the agent or session state and the returned interrupt IDs as described in Interrupts.