The Agent Guide — StreamFlow Pulse
Observe → decide → act. An agent is a pipeline on a stream — not a chat loop.
Table of Contents
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
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.outScaffolding 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
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.
| Capability | What you declare |
|---|---|
| Windows | tumbling / sliding, keyed by any field, event-time with watermarks; late events route to a DLQ |
| Aggregations | count · sum · min · max · avg · first · last · stddev · countWhere(expr) · sumWhere(expr) |
| Operators | filter · map · reduce · aggregate · dedup · join (stream-stream) · process (keyed state + timers) |
| Inputs | multi-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 default — pulse 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:
| Need | Endpoint |
|---|---|
| Send an event in | POST /api/pulse/x/<app>/in |
| Read results | GET /api/pulse/events/<topic> |
| Stream results live | GET /api/pulse/events/stream (SSE) |
| Request/response over a stream | pulse-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
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 agent8. Secrets & configuration
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
| Command | What it does |
|---|---|
| pulse server start --dev | Boot 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-templates | Explore the palette |
| pulse deploy . | Deploy the app: source, stages, sink, API + SDK |
| pulse events send --topic <t> --data @f.json | Publish a test event |
| pulse events tail --topic <t> | Live-tail any topic |
| pulse logs -f | Tail 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.