The Event Pipeline

The fundamentals course introduced the Event Bus as the glue between CAO's components. Now let's look at what actually flows through it — and the five specialized services that make message delivery feel instant.

Think of it as a newsroom. Reporters (FIFOs) feed raw stories into a wire service (EventBus), editors (StatusMonitor) decide what is breaking news, and the dispatch desk (InboxService) routes the right story to the right person at the right time.

Watch a Chunk Travel

Follow a single piece of terminal output as it flows through every stage of the pipeline. Notice how the EventBus appears twice -- once routing raw output, once routing the derived status change. This two-phase publish is the key to decoupling detection from delivery.

The Sticky Latch

StatusMonitor uses debounced edge detection -- it does not re-evaluate status on every single chunk. Instead it waits for output to settle, then checks. And once it determines a terminal is IDLE, it applies a "sticky latch":

Here is the core loop, simplified. The real code adds debouncing, buffer rotation, and the latch logic, but the publish/subscribe pattern remains the same:

# services/status_monitor.py (simplified)
queue = bus.subscribe("terminal.*.output")  # wildcard!
async for topic, data in queue:
    terminal_id = topic.split(".")[1]
    buffer[terminal_id] += data
    status = provider.get_status(buffer[terminal_id])
    if status != last_status[terminal_id]:
        bus.publish(f"terminal.{terminal_id}.status", {"status": status})

Why Workflows?

The supervisor pattern with assign() is powerful — it lets a creative agent decompose problems on the fly, delegate to workers in parallel, and adapt to unexpected results. But sometimes you do not want creativity. You want a fixed pipeline that runs the same steps, in the same order, every time.

Workflows give you exactly that: a declarative YAML spec that defines each step, its inputs, its outputs, and how data flows between them. No improvisation, no skipped steps, no variation between runs.

Anatomy of a Workflow

A workflow spec is a flat YAML file with top-level metadata and a list of steps. Specs live in ~/.aws/cli-agent-orchestrator/workflows/.

name: code-review-pipeline
description: Automated PR review pipeline
inputs:
  pr_url:
    type: string
    required: true
steps:
  - id: fetch-diff
    provider: claude_code
    agent: developer
    prompt: "Fetch the diff from {{workflow.inputs.pr_url}} and summarize"
    output_schema:
      type: object
      properties:
        summary: {type: string}

  - id: review-security
    provider: claude_code
    agent: reviewer
    prompt: "Review for security issues: {{steps.fetch-diff.output.summary}}"

  - id: review-quality
    provider: kiro_cli
    agent: reviewer
    prompt: "Review for code quality: {{steps.fetch-diff.output.summary}}"

Running a Workflow

From authoring to monitoring — the lifecycle of a workflow run:

Check Your Understanding

Skills

A skill is a folder containing a SKILL.md file with structured instructions that an agent can load at runtime. Skills live in ~/.aws/cli-agent-orchestrator/skills/ and are managed via the CLI: cao skills add, cao skills remove, cao skills list.

When an agent needs a skill, it calls the load_skill(name) MCP tool. For Kiro CLI, skills are delivered natively via skill:// resources in the agent JSON. Kiro CLI and OpenCode CLI discover skills natively; other providers load them on demand through the MCP server.

Installing Agents

The cao install command takes a profile name, local path, or URL and sets it up for your provider automatically. No manual JSON editing or config wrangling required.

# Install from built-in registry
cao install code_supervisor

# Install from a local file
cao install ./my-agent.md

# Install from URL with env var
cao install https://raw.githubusercontent.com/org/repo/main/agent.md \
  --env TEAM=platform

# Manage environment variables
cao env set ANTHROPIC_API_KEY sk-...
cao env list
cao env unset ANTHROPIC_API_KEY

Scheduled Flows

Scheduled Flows let you run agent tasks on a cron-like schedule without any human trigger. Define a flow as a Markdown file with YAML frontmatter, register it with the CLI, and CAO's background daemon handles the rest.

---
name: nightly-review
schedule: "0 2 * * *"
agent_profile: code_supervisor
provider: claude_code
---

Review all PRs opened today and
summarize findings.

The server's background daemon checks every 60 seconds for flows whose cron expression matches. Sessions are auto-named cao-flow-{name}.

CLI commands: cao schedule add FILE, cao schedule list, cao schedule run NAME, cao schedule enable/disable NAME, cao schedule remove NAME.

Flows can include a script step that runs first and outputs JSON. If it returns {"execute": false}, the flow is skipped for that cycle — useful for conditional execution based on external state.

The Ops MCP Server

CAO ships two MCP servers with very different purposes. Understanding which is which is essential for building integrations.

Ops server tools:

• list_profiles — discover available agent profiles
• install_profile — install a profile by name or URL
• launch_session — start a new CAO session
• send_session_message — deliver a message to a running session
• get_terminal_status — check if a terminal is idle, processing, etc.
• get_terminal_output — read recent terminal output
• read_session_output — read a terminal's captured output by session name
• get_profile_details — get detailed information about a specific profile
• list_sessions — list all active CAO sessions
• get_session_info — get detailed info about a specific session
• shutdown_session — gracefully stop a session

Fully Autonomous Operation

When an agent hits a permission prompt or asks for clarification, it enters WAITING_USER_ANSWER status. Normally a human would respond. But the answer_user_prompt tool lets another agent respond programmatically.

A supervisor (or another internal agent via the cao-mcp-server) calls answer_user_prompt(terminal_id, answer) to unblock the waiting worker — no human needed. Note: answer_user_prompt is available only through the internal cao-mcp-server, not the ops server. External scripts can send input to a waiting terminal via POST /terminals/{id}/input — the same generic endpoint used for all terminal input.

Check Your Understanding

← Back to the fundamentals course