Back to Guides

    The Agent Guide — StreamFlow Pulse

    Observe → decide → act. An agent is a pipeline on a stream — not a chat loop.

    1. The mental model

    An agent in Pulse is not a chat loop. It is a pipeline that lives on a stream: a source feeds events into a topic, a chain of stages transforms and decides, and a sink (or the next agent) receives what comes out. Each stage runs one of four engines and publishes its output to its own topic — <app>.in → <stage>.out → … — so every intermediate decision is observable, tailable and replayable.

    The loop to keep in your head: observe → decide → act. Sources observe. Stages decide. Sinks and MCP tools act. Because every output is just another topic, "multi-agent" is not a framework feature — it's pointing the next agent's input at the previous agent's output.

    2. Quick start — four commands

    bash
    1# 0. boot a local Pulse and log the CLI in (dev mode)
    2pulse server start --dev
    3
    4# 1. scaffold: source → stages → sink
    5pulse new oncall \
    6  --source file-tail \
    7  --stage rule-based:triage \
    8  --stage llm:reason \
    9  --stage mcp:act \
    10  --sink webhook
    11
    12# 2. fill in pulse.yaml (paths, rules, prompt, tools) then deploy
    13pulse deploy .
    14
    15# 3. feed it an event and watch it decide, live
    16pulse events send --topic oncall.in --data @sample.json
    17pulse events tail --topic oncall.reason.out

    Scaffolding writes three files: pulse.yaml (the app), sample.json (a test event) and a README.md. Deploy stands up the source, the stage pipeline chained by topics, the sink — plus a typed HTTP API and SDK for the app.

    Discovering what's available: pulse new --list-sources, --list-sinks and --list-operators enumerate the connector and operator palette. --list-templates shows ready-made apps you can start from with --from-template.

    3. Anatomy of pulse.yaml

    pulse.yaml
    1source:                    # OBSERVE — where events come from
    2  kind: file-tail          # file-tail · webhook · http-poll · jdbc-source · …
    3  path: /var/log/app/events.log
    4
    5stages:                    # DECIDE — each stage = one engine
    6  - name: triage
    7    engine: rule-based
    8    rules: [ "severity == 'error'" ]
    9
    10  - name: reason
    11    engine: llm
    12    systemPrompt: |
    13      Classify the incident and draft a
    14      one-line on-call summary.
    15
    16  - name: act
    17    engine: mcp
    18    mcpTools: [ pagerduty.createIncident ]
    19
    20sink:                      # ACT — where results go (optional)
    21  kind: webhook
    22  url: ${secret:ALERT_WEBHOOK}
    • source — one block, one connector kind. Its events land on <app>.in.
    • stages — an ordered chain. Stage n consumes stage n−1's output topic and publishes to <app>.<stage>.out.
    • sink — optional. Anything the palette offers: webhook, chat, JDBC, Kafka-protocol, files…
    • secrets — always $${secret:NAME} or $${env:VAR}, never inline. Deploy fails loudly on unresolved references instead of shipping a broken config.

    4. The four engines

    The design rule that keeps agents fast and cheap: use the cheapest engine that can make each decision. Shape with streaming, gate with rule-based, reserve llm for genuine ambiguity, and reach into the world with mcp.

    engine: streaming — shape the stream

    Deterministic, stateful stream processing at microsecond cost. No model anywhere in the path.

    CapabilityWhat you declare
    Windowstumbling / sliding, keyed by any field, event-time with watermarks; late events route to a DLQ
    Aggregationscount · sum · min · max · avg · first · last · stddev · countWhere(expr) · sumWhere(expr)
    Operatorsfilter · map · reduce · aggregate · dedup · join (stream-stream) · process (keyed state + timers)
    Inputsmulti-topic union — one stage can consume several topics

    engine: rule-based — decide the obvious

    Declarative predicates over event fields. Events that match pass through (optionally transformed); the rest stop. Use it to keep 95% of traffic away from the LLM: routing, gating, thresholds, triage.

    engine: llm — reason over the hard ones

    A model call per event, driven by your systemPrompt. Bring your own provider: configure a hosted model API key, or point at a local model runtime for on-prem inference. The stage receives the event (plus upstream enrichments) and emits the model's structured answer to its output topic.

    Honest constraint: no provider configured → the llm stage reports itself unhealthy rather than silently passing events through. Same philosophy everywhere: a broken stage is visible, never invisible.

    engine: mcp — act on the world

    Stages that call real tools over the Model Context Protocol: query systems, create tickets, run workflows, or pause for a human approval — the human is a pipeline step, not an afterthought. Tools come from MCP plugins you install into Pulse; mcpTools lists which ones the stage may invoke.

    Honest constraint: until its plugin is installed, an mcp stage shows blocked in the pipeline view. That's by design — you see exactly what's missing instead of a silent no-op.

    5. Composing multi-agent systems

    There is no special orchestration layer to learn. Topics are the composition primitive:

    • Chain — agent B's source reads agent A's output topic. A pipeline of pipelines.
    • Fan out — several agents subscribe to the same topic; each sees every event.
    • Fan in — one streaming stage takes a multi-topic union and merges flows.
    • Correlate — a stream-stream join matches events across two flows inside a window (an order and its payment, an alert and its ack).

    Because every hop is a durable topic, the seams between agents are observable by defaultpulse events tail works at every joint, not just at the ends.

    6. Calling your agent from outside

    Every deployed app exposes the same HTTP surface, so any client — a script, a workflow engine, an orchestrator — integrates the same way:

    NeedEndpoint
    Send an event inPOST /api/pulse/x/<app>/in
    Read resultsGET /api/pulse/events/<topic>
    Stream results liveGET /api/pulse/events/stream (SSE)
    Request/response over a streampulse-py client.duplex() (WebSocket corrélé)

    Worked bridge examples ship in the repo for the most common workflow engines, orchestrators and integration platforms — the same Pulse surface each time, only the client node differs.

    7. Durability, replay & debugging

    • Crash-resume. Per-agent stream offsets are persisted. Restart Pulse mid-flight and every agent resumes exactly where it stopped — no gap, no double-processing.
    • Deterministic replay. Re-run an agent over the exact events it saw and get byte-identical decisions in mocked mode — the tool for "why did it do that?" investigations.
    • Windowed-state recovery. Checkpointed window state survives a kill and rebuilds on restart; long-running aggregates aren't lost to one bad process.
    • Per-event trajectory. Each event's path through the stages is traced (W3C-compatible) and persisted — you can time-travel a single event through the pipeline after the fact.
    • Dead-letter queues. Failing events and genuinely-late window arrivals land in a durable DLQ instead of vanishing.

    The everyday debugging loop

    bash
    1pulse logs -f                                 # tail the engine log
    2pulse events tail --topic oncall.reason.out   # watch a stage's output live
    3pulse metrics agent reason                    # throughput / lag / errors per agent

    8. Secrets & configuration

    bash
    1pulse secret set ALERT_WEBHOOK https://hooks.example.com/…   # stored locally, 0600
    • $${secret:NAME} resolves from the CLI's local secret store at deploy time.
    • $${env:VAR} resolves from the environment.
    • Unresolved = failed deploy. A reference that can't be resolved aborts the deploy with a named error — a config with a hole in it never reaches production.

    9. CLI quick reference

    CommandWhat it does
    pulse server start --devBoot a local Pulse and auto-login the CLI
    pulse new <app> --source … --stage … --sink …Scaffold pulse.yaml + sample.json + README
    pulse new --list-sources / --list-sinks / --list-operators / --list-templatesExplore the palette
    pulse deploy .Deploy the app: source, stages, sink, API + SDK
    pulse events send --topic <t> --data @f.jsonPublish a test event
    pulse events tail --topic <t>Live-tail any topic
    pulse logs -fTail the engine log
    pulse metrics agent <name>Per-agent throughput, lag, errors
    pulse secret set <NAME> <value>Store a secret for ${secret:NAME}

    Now build one.

    The fastest way to internalize the model is to watch an agent decide on live events — it takes about three minutes.