Skip to content

Workflows Guide#

YAML workflows provide a declarative way to define multi-agent coordination without writing code. This guide walks through creating, running, and customizing workflows.

Quick Start#

1. Create a Workflow File#

Create my_workflow.yaml:

openintent: "1.0"

info:
  name: "My First Workflow"
  description: "A simple two-phase workflow"

workflow:
  research:
    title: "Research Topic"
    description: "Gather information about the topic"
    assign: researcher

  summarize:
    title: "Create Summary"
    description: "Summarize the research findings"
    assign: summarizer
    depends_on: [research]

2. Start the Server#

openintent-server

3. Run Your Agents#

Each agent referenced in the workflow needs a running implementation:

# researcher.py
from openintent import Agent, on_assignment

@Agent("researcher")
class ResearchAgent:
    @on_assignment
    async def handle(self, intent):
        return {"findings": "Research results here..."}

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

# summarizer.py
from openintent import Agent, on_assignment

@Agent("summarizer")
class SummarizerAgent:
    @on_assignment
    async def handle(self, intent):
        findings = intent.state.get("findings", "")
        return {"summary": f"Summary of: {findings[:100]}..."}

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

4. Execute the Workflow#

openintent run my_workflow.yaml

Workflow Structure#

Minimal Workflow#

The smallest valid workflow:

openintent: "1.0"

info:
  name: "Minimal"

workflow:
  task:
    assign: my-agent

Full Workflow#

A workflow using all features:

openintent: "1.0"

info:
  name: "Full Example"
  version: "2.0.0"
  description: "Demonstrates all workflow features"

governance:
  max_cost_usd: 10.00
  timeout_hours: 24
  require_approval:
    when: "risk == 'high'"
    approvers: [admin]

agents:
  researcher:
    description: "Gathers information"
    capabilities: [search, analysis]

llm:
  provider: openai
  model: gpt-5.2

workflow:
  research:
    title: "Research Phase"
    assign: researcher
    constraints:
      - "Be thorough"
    initial_state:
      depth: "comprehensive"
    retry:
      max_attempts: 3
      backoff: exponential
    outputs:
      - findings

  synthesis:
    title: "Synthesis Phase"
    assign: synthesizer
    depends_on: [research]
    cost_tracking:
      enabled: true
      budget_usd: 5.00

Dependencies#

Sequential Dependencies#

Phases run in order:

workflow:
  step1:
    assign: agent-a

  step2:
    assign: agent-b
    depends_on: [step1]

  step3:
    assign: agent-c
    depends_on: [step2]

Parallel Execution#

Phases without dependencies run in parallel:

workflow:
  # These three run simultaneously
  research_a:
    assign: researcher
  research_b:
    assign: researcher
  research_c:
    assign: researcher

  # This waits for all three
  merge:
    assign: synthesizer
    depends_on: [research_a, research_b, research_c]

Diamond Dependencies#

Multiple phases can depend on the same prerequisite:

workflow:
  fetch_data:
    assign: fetcher

  analyze_trends:
    assign: analyst
    depends_on: [fetch_data]

  analyze_sentiment:
    assign: analyst
    depends_on: [fetch_data]

  report:
    assign: reporter
    depends_on: [analyze_trends, analyze_sentiment]

Passing Data Between Phases#

Using State#

Each phase can read state from previous phases:

workflow:
  research:
    assign: researcher
    initial_state:
      topic: "AI trends"
    outputs:
      - findings

  summarize:
    assign: summarizer
    depends_on: [research]
    # Agent can access intent.state.findings

In your agent:

@Agent("summarizer")
class SummarizerAgent:
    @on_assignment
    async def handle(self, intent):
        # Access state from previous phase
        findings = intent.state.get("findings", [])

        summary = create_summary(findings)
        return {"summary": summary}

Using Constraints#

Pass requirements to agents:

workflow:
  research:
    assign: researcher
    constraints:
      - "Use at least 3 sources"
      - "Focus on peer-reviewed content"
      - "Include publication dates"

In your agent:

@Agent("researcher")
class ResearchAgent:
    @on_assignment
    async def handle(self, intent):
        # Constraints available in intent.constraints
        for constraint in intent.constraints:
            print(f"Must satisfy: {constraint}")

Governance#

Cost Budgets#

Limit total workflow cost:

governance:
  max_cost_usd: 10.00

Per-phase budgets:

workflow:
  expensive_task:
    assign: my-agent
    cost_tracking:
      enabled: true
      budget_usd: 5.00
      alert_at_percent: 80

Timeouts#

Set workflow timeout:

governance:
  timeout_hours: 24

Approval Gates#

Require human approval:

governance:
  require_approval:
    when: "risk.level == 'high'"
    approvers:
      - "manager"
      - "compliance"
    timeout_hours: 4
    on_timeout: "escalate"

Escalation#

Configure escalation contacts:

governance:
  escalation:
    contact: "admin@company.com"
    after_hours: 4

Permissions (RFC-0011)#

Control who can see and modify each phase's data using the unified permissions field.

Quick Start#

The simplest forms — a single string or a list of agents:

workflow:
  public_step:
    assign: researcher
    permissions: open           # Anyone can access

  private_step:
    assign: analyst
    permissions: private        # Only the assigned agent

  team_step:
    assign: lead
    permissions: [analyst, auditor]  # Only these agents

Full Permissions Object#

For fine-grained control, use the full object form with policy, allow, delegate, and context:

workflow:
  sensitive_analysis:
    assign: analyst
    permissions:
      policy: restricted
      default: read
      allow:
        - agent: "analyst"
          level: write
        - agent: "auditor"
          level: read
      delegate:
        to: ["specialist-bot", "backup-bot"]
        level: write
      context: [dependencies, peers, acl, delegated_by]

Delegation#

When delegate is specified, agents can hand off work:

@Agent("triage-bot")
class TriageAgent:
    @on_assignment
    async def handle(self, intent):
        if needs_specialist(intent):
            await self.delegate(intent.id, "specialist-bot")
            return {"status": "delegated"}
        return {"result": process(intent)}

    @on_access_requested
    async def policy(self, intent, request):
        if "compliance" in (request.capabilities or []):
            return "approve"
        return "defer"

Context Injection#

The context field inside permissions controls what context the agent automatically receives. In your agent, context is available via intent.ctx:

@Agent("analyst")
class AnalystAgent:
    @on_assignment
    async def handle(self, intent):
        ctx = intent.ctx
        if ctx.delegated_by:
            print(f"Delegated by: {ctx.delegated_by}")

        for dep_id, dep_state in ctx.dependencies.items():
            print(f"Dependency {dep_id}: {dep_state}")

        return {"analysis": "complete"}

Governance Access Review#

Set a workflow-level policy for access requests:

governance:
  access_review:
    on_request: defer
    approvers: ["admin-agent", "compliance-officer"]
    timeout_hours: 4

Backward Compatibility#

Legacy access, delegation, and context fields at the phase level are still parsed and auto-converted to the unified permissions format. New workflows should always use permissions.

Retry Policies#

Handle transient failures:

workflow:
  api_call:
    assign: api-agent
    retry:
      max_attempts: 5
      backoff: exponential       # constant, linear, exponential
      initial_delay_ms: 1000
      max_delay_ms: 60000
      retryable_errors:
        - "TIMEOUT"
        - "RATE_LIMIT"
        - "503"
      fallback_agent: "backup-agent"

Leasing#

Prevent concurrent access conflicts:

workflow:
  update_database:
    assign: db-agent
    leasing:
      strategy: global           # global, per_section, custom
      scope: "database:users"
      ttl_seconds: 300
      wait_for_lock: true

Attachments#

Declare expected file outputs:

workflow:
  generate_report:
    assign: reporter
    attachments:
      - filename: "report.pdf"
        content_type: "application/pdf"
      - filename: "data.json"
        content_type: "application/json"

LLM Configuration#

Configure the default LLM provider:

llm:
  provider: openai              # openai, anthropic, env
  model: gpt-5.2
  temperature: 0.7
  max_tokens: 4096
  system_prompt: "You are a helpful research assistant."

Use provider: env to auto-detect from environment variables:

llm:
  provider: env                 # Uses OPENAI_API_KEY or ANTHROPIC_API_KEY

CLI Commands#

Run a Workflow#

openintent run workflow.yaml

Options:

--server URL        # OpenIntent server URL (default: http://localhost:8000)
--api-key KEY       # API key for authentication
--timeout SECONDS   # Execution timeout (default: 300)
--output FILE       # Save results to JSON file
--dry-run           # Validate without executing
--verbose           # Show detailed progress

Validate a Workflow#

openintent validate workflow.yaml

List Sample Workflows#

openintent list

Create from Template#

openintent new "My Workflow Name"

Programmatic Usage#

Load and Execute#

from openintent.workflow import WorkflowSpec

# Load from file
spec = WorkflowSpec.from_yaml("workflow.yaml")

# Execute
result = await spec.run(
    server_url="http://localhost:8000",
    api_key="dev-user-key",
    timeout=300,
    verbose=True
)

Load from String#

yaml_content = """
openintent: "1.0"
info:
  name: "Dynamic Workflow"
workflow:
  task:
    assign: my-agent
"""

spec = WorkflowSpec.from_string(yaml_content)

Convert to PortfolioSpec#

For manual execution with a Coordinator:

from openintent.workflow import WorkflowSpec
from openintent import Coordinator

spec = WorkflowSpec.from_yaml("workflow.yaml")
portfolio_spec = spec.to_portfolio_spec()

coordinator = Coordinator("orchestrator", base_url, api_key)
result = await coordinator.execute(portfolio_spec)

Validation#

from openintent.workflow import validate_workflow, WorkflowValidationError

try:
    warnings = validate_workflow("workflow.yaml")
    for warning in warnings:
        print(f"Warning: {warning}")
except WorkflowValidationError as e:
    print(f"Error: {e}")

Best Practices#

1. Start Simple#

Begin with a minimal workflow and add complexity incrementally:

openintent: "1.0"
info:
  name: "Simple Start"
workflow:
  main_task:
    assign: my-agent

2. Use Descriptive Names#

Phase names become intent titles. Make them clear:

workflow:
  # Good
  gather_customer_feedback:
    assign: feedback-agent

  # Less clear
  step1:
    assign: agent

3. Document Agents with Capabilities#

Declare agents with capabilities for access decisions:

agents:
  researcher:
    description: "Searches and analyzes web sources"
    capabilities: [search, summarization]
    default_permission: read

4. Set Reasonable Budgets#

Always set cost limits for production:

governance:
  max_cost_usd: 25.00

workflow:
  expensive_analysis:
    assign: analyst
    cost_tracking:
      enabled: true
      budget_usd: 10.00

5. Use Retry Policies#

Handle transient failures gracefully:

workflow:
  api_dependent_task:
    assign: api-agent
    retry:
      max_attempts: 3
      backoff: exponential
      retryable_errors: ["TIMEOUT", "RATE_LIMIT"]

6. Define Outputs#

Explicitly declare what each phase produces:

workflow:
  research:
    assign: researcher
    outputs:
      - sources
      - findings
      - confidence_score

Troubleshooting#

"Missing 'openintent' version field"#

Add the version at the top:

openintent: "1.0"

"Phase 'x' missing 'assign' field"#

Every phase needs an agent:

workflow:
  my_phase:
    assign: my-agent  # Required

"Circular dependency detected"#

Remove one dependency to break the cycle:

# Invalid
workflow:
  a:
    depends_on: [c]
  b:
    depends_on: [a]
  c:
    depends_on: [b]  # Creates cycle: a -> c -> b -> a

"Unknown phase referenced"#

Check spelling in depends_on:

workflow:
  research:
    assign: researcher

  synthesis:
    depends_on: [reserch]  # Typo! Should be 'research'

See Also#