Workflows

A workflow is a named unit of work. It owns payload validation, prompt construction, step sequencing, and result shaping. Callers know only its name — see Invocation contract.

Types

WorkflowDefinition protocol (agentcore/workflows/registry.py):

@runtime_checkable
class WorkflowDefinition(Protocol):
    name: str
    description: str
    def run(self, payload: dict, options: RunOptions | None = None) -> WorkflowResult: ...

That is the entire requirement. Extending BaseWorkflow is optional and only useful for multi-step workflows.

@dataclass
class WorkflowStep:
    name: str                                  # unique slug; used as a dependency key
    prompt: str                                # supports {{variable}} placeholders
    depends_on: list[str] = field(default_factory=list)
    options: RunOptions | None = None           # per-step overrides
    output_schema: Any | None = None             # Pydantic BaseModel subclass

@dataclass
class WorkflowStepResult:
    name: str
    output: str | None                          # None when the step failed
    structured_output: Any = None
    usage: dict[str, int] | None = None
    error: str | None = None

@dataclass
class WorkflowResult:
    success: bool
    output: str
    structured_output: Any = None
    stop_reason: str | None = None
    usage: dict[str, int] | None = None
    error: str | None = None
    steps: list[WorkflowStepResult] | None = None

`WorkflowRegistry`

class WorkflowRegistry:
    def register(self, *workflows: WorkflowDefinition) -> WorkflowRegistry   # chainable
    def list(self) -> list[str]
    def run(self, name, payload=None, options=None) -> WorkflowResult

register() raises ValueError on a duplicate name — a collision fails startup.
run() raises KeyError with a message listing the available workflows when the name is unknown.
One registry per process, built once by build_registry().

`BaseWorkflow`

agentcore/workflows/base_workflow.py. Register steps in the constructor, then call execute_steps() from run().

class MyWorkflow(BaseWorkflow):
    name = "my-workflow"
    description = "Does the thing."

    def __init__(self) -> None:
        super().__init__(my_agent)
        self.add_step(WorkflowStep(name="fetch", prompt="Fetch data for {{id}}."))
        self.add_step(WorkflowStep(
            name="analyse", prompt="Analyse the data above.",
            depends_on=["fetch"], output_schema=AnalysisOutput,
        ))

    def run(self, payload, options=None):
        if not payload.get("id"):
            return WorkflowResult(success=False, output="", error="id is required")
        return self.execute_steps({"id": payload["id"]}, options)

`add_step()`

Chainable. Validates that every name in depends_on refers to an already registered step, so forward references raise immediately at construction time — which means at container startup, not at request time.

`execute_steps()`

Steps run in registration order against a single Strands Agent obtained from self._agent.new_session(options). Because it is one Agent, agent.messages accumulates across steps and each step sees the whole prior conversation with no manual context threading.

Per step:

Verify every depends_on step has completed. Any missing one fails the workflow immediately with Missing dependencies: ….
Substitute {{variable}} placeholders from variables.
When output_schema is set, append a JSON-schema instruction to the prompt (derived from model_json_schema()) demanding a single JSON object with no markdown fences or prose.
Call agent(resolved_prompt).
When output_schema is set, strip any markdown fences and validate with model_validate_json(). Validation failure is swallowed — the raw text is kept and structured_output stays None.
Accumulate token usage and append a WorkflowStepResult.
On an exception, stop immediately and return success=False naming the failing step.

`_generate_prompt()`

re.sub(r"\{\{\s*([\w]+)\s*\}\}", …) against the variables dict. Unknown keys substitute to an empty string rather than raising — a missing variable produces a degraded prompt, not a crash.

Structured output

Two independent mechanisms, and it is worth knowing which is in play.

WorkflowStep.output_schema (this module): BaseWorkflow appends a schema instruction to the prompt text and validates the response itself. Prompt-level, best-effort, tolerant of failure.
AgentConfig.structured_output_model (morpheus-llm): Strands adds a dedicated output tool, so the model is structurally constrained and AgentResult.structured_output carries a validated instance. Stronger. Used by the Lambda agents. See structured output.

The dialler workflow uses the first; the enrichment, scoring, and email categorisation agents use the second.

Built-in workflows

`PromptWorkflow`

agentcore/workflows/builtin/prompt_workflow.py. Implements the protocol directly — no BaseWorkflow. Takes an agent, a name, and a description at construction, so one class serves any agent:

PromptWorkflow(dialler_agent, "test", "Sample test workflow")

Validates payload.prompt, calls self._agent.run(prompt, options), and wraps the result. Exceptions become WorkflowResult(success=False, error=str(exc)).

`DiallerAgentWorkflow`

agentcore/workflows/builtin/dialler_workflow.py. Two steps.

Step 1 — analyse-history

Prompt: get contact information for {{phoneNumber}} from HubSpot; retrieve all previous call history, transcripts, emails and other interactions; understand the content and outcome of each; identify the last interaction, open tasks, and the current relationship status.

class AnalyseHistoryOutput(BaseModel):
    contact: _ContactInfo                       # name, email, phone, company,
                                                # hubspot_id, lifecycle_stage, hubspot_url
    last_interaction: _LastInteraction | None    # date, type, summary
    open_tasks: list[str] | None
    relationship_status: str
    interaction_history: list[_InteractionHistoryItem] | None   # date, type, summary

Step 2 — prepare-brief

The tool work all happens in step 1 through the HubSpot MCP client; step 2 is pure reasoning over the conversation the agent already has in context.

`build_registry()`

def build_registry() -> WorkflowRegistry:
    return WorkflowRegistry().register(
        PromptWorkflow(dialler_agent, "test", "Sample test workflow"),
        DiallerAgentWorkflow(),
    )

Takes no arguments — agents and tools are self-contained module-level singletons. Every new workflow must be added here or it is unreachable.

Adding a workflow

Create agentcore/workflows/builtin/<name>_workflow.py.
Validate required payload keys at the top of run() and return WorkflowResult(success=False, …) for anything missing.
Extend BaseWorkflow for multi-step; implement the protocol directly for single-step.
Register it in agentcore/workflows/init.py.
Document the name and payload in Invocation contract.

Guidelines

Use descriptive step slugs. They appear in WorkflowResult.steps and in error messages: lookup-contact, not step1.
One task per step. If a step’s prompt has two verbs, it is two steps.
Pass runtime data through variables, not by string-concatenating into prompts. {{placeholder}} substitution keeps the prompt readable and diffable.
Declare depends_on honestly. It is the only thing that guards against a step running before the context it needs exists.
Add an output_schema to any step whose output is consumed programmatically — and handle structured_output is None, because validation failure is non-fatal.