Skip to content

Workflow API Reference#

Python API for loading, validating, and executing YAML workflows.

WorkflowSpec#

The main class for working with workflow YAML files.

WorkflowSpec dataclass #

WorkflowSpec(
    version: str,
    name: str,
    description: str = "",
    workflow_version: str = "1.0.0",
    agents: dict[str, dict[str, Any]] = dict(),
    phases: list[PhaseConfig] = list(),
    governance: Optional[GovernanceConfig] = None,
    llm: Optional[LLMConfig] = None,
    identity: Optional[IdentityConfig] = None,
    verification: Optional[VerificationConfig] = None,
    types: dict[str, Any] = dict(),
    source_path: Optional[Path] = None,
)

Parsed and validated workflow specification.

Load from YAML

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

Convert to PortfolioSpec

portfolio = spec.to_portfolio_spec()

Run directly

result = await spec.run(server_url="http://localhost:8000")

from_yaml classmethod #

from_yaml(path: str | Path) -> 'WorkflowSpec'

Load and validate a workflow from a YAML file.

Parameters:

Name Type Description Default
path str | Path

Path to the YAML file

 required

Returns:

Type Description
'WorkflowSpec'

Validated WorkflowSpec

Raises:

Type Description
WorkflowNotFoundError

File not found
WorkflowValidationError

Invalid YAML structure

from_string classmethod #

from_string(
    yaml_content: str, source_name: str = "<string>"
) -> "WorkflowSpec"

Load and validate a workflow from a YAML string.

Parameters:

Name Type Description Default
yaml_content str

YAML content as a string

 required
source_name str

Name for error messages

 '<string>'

Returns:

Type Description
'WorkflowSpec'

Validated WorkflowSpec

to_portfolio_spec #

to_portfolio_spec() -> 'PortfolioSpec'

Convert this workflow to a PortfolioSpec for execution.

Returns:

Type Description
'PortfolioSpec'

PortfolioSpec ready for execution

run async #

run(
    server_url: str = "http://localhost:8000",
    api_key: str = "dev-user-key",
    timeout: int = 300,
    verbose: bool = True,
) -> dict

Execute this workflow against an OpenIntent server.

Parameters:

Name Type Description Default
server_url str

OpenIntent server URL

 'http://localhost:8000'
api_key str

API key for authentication

 'dev-user-key'
timeout int

Execution timeout in seconds

 300
verbose bool

Print progress messages

 True

Returns:

Type Description
dict

Workflow execution result

Example Usage#

from openintent.workflow import WorkflowSpec

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

# Access metadata
print(f"Workflow: {spec.name}")
print(f"Version: {spec.workflow_version}")
print(f"Phases: {len(spec.phases)}")

# List phases
for phase in spec.phases:
    print(f"  - {phase.title} -> {phase.assign}")

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

Attributes#

Attribute Type Description
version str Protocol version (e.g., "1.0")
name str Workflow name
description str Workflow description
workflow_version str Semantic version
agents dict[str, dict] Declared agents
phases list[PhaseConfig] Workflow phases
governance GovernanceConfig | None Governance settings
llm LLMConfig | None LLM configuration
types dict[str, Any] Type definitions
source_path Path | None Source file path

PhaseConfig#

Configuration for a single workflow phase.

PhaseConfig dataclass #

PhaseConfig(
    name: str,
    title: str,
    assign: str,
    description: str = "",
    depends_on: list[str] = list(),
    constraints: list[str] = list(),
    initial_state: dict[str, Any] = dict(),
    retry: Optional[dict[str, Any]] = None,
    leasing: Optional[dict[str, Any]] = None,
    cost_tracking: Optional[dict[str, Any]] = None,
    attachments: Optional[list[dict[str, Any]]] = None,
    permissions: Optional[PermissionsConfig] = None,
    inputs: dict[str, str] = dict(),
    outputs: list[str] = list(),
    skip_when: Optional[str] = None,
)

Configuration for a workflow phase (intent).

Attributes#

Attribute Type Description
name str Internal phase name (from YAML key)
title str Display title
assign str Agent ID to assign
description str Phase description
depends_on list[str] Dependency phase names
constraints list[str] Constraints for the agent
initial_state dict[str, Any] Initial state values
retry dict | None Retry policy (RFC-0010)
leasing dict | None Leasing config (RFC-0003)
cost_tracking dict | None Cost tracking (RFC-0009)
attachments list[dict] | None Attachments (RFC-0005)
inputs dict[str, str] Input mappings
outputs list[str] Output state keys
skip_when str | None Skip condition

Example#

for phase in spec.phases:
    print(f"Phase: {phase.title}")
    print(f"  Assigned to: {phase.assign}")
    print(f"  Depends on: {phase.depends_on}")
    print(f"  Constraints: {phase.constraints}")

    if phase.retry:
        print(f"  Retry: {phase.retry.get('max_attempts')} attempts")

LLMConfig#

LLM provider configuration.

LLMConfig dataclass #

LLMConfig(
    provider: str = "openai",
    model: str = "",
    temperature: float = 0.7,
    max_tokens: int = 4096,
    system_prompt: str = "",
)

LLM provider configuration for the workflow.

get_default_model #

get_default_model() -> str

Get the default model for this provider.

get_env_key #

get_env_key() -> str

Get the environment variable key for this provider.

Attributes#

Attribute Type Default Description
provider str "openai" Provider name
model str "" Model identifier
temperature float 0.7 Sampling temperature
max_tokens int 4096 Maximum tokens
system_prompt str "" System prompt

Methods#

get_env_key()#

Returns the environment variable name for the API key.

config = spec.llm
print(config.get_env_key())  # "OPENAI_API_KEY"

get_default_model()#

Returns the default model for the provider.

config = LLMConfig(provider="anthropic")
print(config.get_default_model())  # "claude-sonnet-4-20250514"

GovernanceConfig#

Governance settings for approval gates and budgets.

GovernanceConfig dataclass #

GovernanceConfig(
    require_approval: Optional[dict[str, Any]] = None,
    max_cost_usd: Optional[float] = None,
    timeout_hours: Optional[float] = None,
    escalation: Optional[dict[str, str]] = None,
    access_review: Optional[dict[str, Any]] = None,
    audit_access_events: bool = True,
)

Governance configuration for the workflow.

Attributes#

Attribute Type Description
require_approval dict | None Approval gate configuration
max_cost_usd float | None Maximum workflow cost
timeout_hours float | None Maximum duration
escalation dict[str, str] | None Escalation contacts

Example#

if spec.governance:
    if spec.governance.max_cost_usd:
        print(f"Budget: ${spec.governance.max_cost_usd}")
    if spec.governance.timeout_hours:
        print(f"Timeout: {spec.governance.timeout_hours}h")

Validation Functions#

validate_workflow()#

Validate a workflow file and return warnings.

validate_workflow #

validate_workflow(path: str | Path) -> list[str]

Validate a workflow file and return any warnings.

Parameters:

Name Type Description Default
path str | Path

Path to workflow YAML

 required

Returns:

Type Description
list[str]

List of warning messages (empty if valid)

Raises:

Type Description
WorkflowValidationError

If workflow is invalid

 
from openintent.workflow import validate_workflow, WorkflowValidationError

try:
    warnings = validate_workflow("workflow.yaml")
    for warning in warnings:
        print(f"Warning: {warning}")
    print("Workflow is valid!")
except WorkflowValidationError as e:
    print(f"Validation failed: {e}")

list_sample_workflows()#

List available sample workflows.

list_sample_workflows #

list_sample_workflows() -> list[dict[str, str]]

List available sample workflows.

Returns:

Type Description
list[dict[str, str]]

List of workflow info dicts with name, description, path

 
from openintent.workflow import list_sample_workflows

for workflow in list_sample_workflows():
    print(f"{workflow['name']}: {workflow['description']}")
    print(f"  Path: {workflow['path']}")
    print(f"  Phases: {workflow['phases']}")

Exceptions#

WorkflowError#

Base exception for workflow errors.

WorkflowError #

Bases: Exception

Base exception for workflow errors.

WorkflowValidationError#

Raised when workflow YAML is invalid.

WorkflowValidationError #

WorkflowValidationError(
    message: str, path: str = "", suggestion: str = ""
)

Bases: WorkflowError

Raised when workflow YAML is invalid.

from openintent.workflow import WorkflowSpec, WorkflowValidationError

try:
    spec = WorkflowSpec.from_yaml("invalid.yaml")
except WorkflowValidationError as e:
    print(f"Path: {e.path}")
    print(f"Error: {e}")
    print(f"Suggestion: {e.suggestion}")

WorkflowNotFoundError#

Raised when workflow file is not found.

WorkflowNotFoundError #

Bases: WorkflowError

Raised when workflow file is not found.

Integration with Agents#

Converting to PortfolioSpec#

For manual execution with a Coordinator:

from openintent.workflow import WorkflowSpec
from openintent import Coordinator

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

# Convert to PortfolioSpec
portfolio_spec = spec.to_portfolio_spec()

# Execute with Coordinator
coordinator = Coordinator(
    agent_id="orchestrator",
    base_url="http://localhost:8000",
    api_key="dev-user-key"
)

result = await coordinator.execute(portfolio_spec)

Direct Execution#

For simpler use cases:

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

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

CLI Reference#

The workflow module provides CLI commands via the openintent command:

# Run a workflow
openintent run workflow.yaml [OPTIONS]

Options:
  --server URL      Server URL (default: http://localhost:8000)
  --api-key KEY     API key
  --timeout SEC     Timeout in seconds (default: 300)
  --output FILE     Save results to file
  --dry-run         Validate only
  --verbose         Show progress

# Validate a workflow
openintent validate workflow.yaml

# List sample workflows
openintent list

# Create from template
openintent new "Workflow Name"

See Also#