StreamFlow Agent Guide

Name: StreamFlow
Author: StreamFlow

Create, Scale, and Orchestrate AI Agents

1.Create a Single Agent
2.Create a Fleet of Agents
3.Create a Graph of Agents (Pipeline)
4.Auto-Scaling
5.Monitor Agents in the UI
6.Agent Lifecycle Management
7.Agent Failover (High Availability)

1. Create a Single Agent

Via REST API

curl

1curl -X POST http://localhost:8080/api/v1/agents \
2  -H "Content-Type: application/json" \
3  -d '{
4    "name": "fraud-detector",
5    "engine": "llm",
6    "inputTopic": "transactions",
7    "outputTopic": "fraud-alerts",
8    "description": "Detects fraudulent transactions using a language model",
9    "instances": 1,
10    "config": {
11      "model": "your-llm-model-id",
12      "systemPrompt": "You are a fraud detection expert...",
13      "temperature": 0.1,
14      "maxTokens": 200
15    }
16  }'

Response

JSON

1{
2  "id": "agent-a1b2c3",
3  "name": "fraud-detector",
4  "status": "RUNNING",
5  "instances": 1,
6  "inputTopic": "transactions",
7  "outputTopic": "fraud-alerts"
8}

Via Java SDK (Programmatic)

Java

1import com.streamflow.sdk.StreamFlowClient;
2import com.streamflow.sdk.StreamFlowProducer;
3import com.streamflow.sdk.StreamFlowConsumer;
4import com.streamflow.sdk.StreamFlowRecord;
5import java.net.URI;
6import java.net.http.*;
7import java.nio.charset.StandardCharsets;
8import java.time.Duration;
9import java.util.List;
10
11public class FraudDetectorExample {
12    public static void main(String[] args) throws Exception {
13
14        // ═══ Step 1: Create the agent via REST API ═══
15        HttpClient http = HttpClient.newHttpClient();
16        String agentJson = """
17            {
18              "name": "fraud-detector",
19              "engine": "rule-based",
20              "inputTopic": "transactions",
21              "outputTopic": "fraud-alerts",
22              "description": "Flags transactions over $10,000",
23              "instances": 1
24            }
25            """;
26
27        HttpRequest createReq = HttpRequest.newBuilder()
28            .uri(URI.create("http://localhost:8080/api/v1/agents"))
29            .header("Content-Type", "application/json")
30            .POST(HttpRequest.BodyPublishers.ofString(agentJson))
31            .build();
32        HttpResponse<String> createResp = http.send(createReq,
33            HttpResponse.BodyHandlers.ofString());
34        System.out.println("Agent created: " + createResp.body());
35
36        // ═══ Step 2: Send transactions ═══
37        StreamFlowClient client = StreamFlowClient.remote("localhost:9092");
38        try (StreamFlowProducer producer = client.producer()) {
39            producer.send("transactions", "tx-001", """
40                {"txId":"tx-001","amount":50.00,"merchant":"Coffee Shop"}
41                """.getBytes(StandardCharsets.UTF_8)).join();
42            producer.send("transactions", "tx-002", """
43                {"txId":"tx-002","amount":15000.00,"merchant":"Crypto Exchange"}
44                """.getBytes(StandardCharsets.UTF_8)).join();
45            producer.send("transactions", "tx-003", """
46                {"txId":"tx-003","amount":120.00,"merchant":"Grocery Store"}
47                """.getBytes(StandardCharsets.UTF_8)).join();
48            producer.flush();
49        }
50        System.out.println("3 transactions sent");
51
52        // ═══ Step 3: Read fraud alerts ═══
53        Thread.sleep(2000);
54        try (StreamFlowConsumer consumer = client.consumer("alert-reader")) {
55            consumer.subscribe("fraud-alerts");
56            List<StreamFlowRecord> alerts = consumer.poll(Duration.ofSeconds(5));
57            System.out.printf("%d fraud alert(s) received:%n", alerts.size());
58            for (StreamFlowRecord alert : alerts) {
59                System.out.printf("  Alert for %s: %s%n",
60                    alert.key(), new String(alert.value()));
61            }
62        }
63    }
64}

Output

1Agent created: {"id":"agent-a1b2c3","name":"fraud-detector","status":"RUNNING"}
23 transactions sent to 'transactions' topic
3
41 fraud alert(s) received:
5  Alert for tx-002: {"txId":"tx-002","fraud":true,"reason":"Amount $15,000 exceeds threshold","confidence":0.95}

How it works

transactions topic          fraud-detector agent         fraud-alerts topic
┌─────────────┐             ┌──────────────────┐         ┌──────────────┐
│ tx-001: $50  │────────────→│  Analyze...      │         │              │
│ tx-002: $15K │────────────→│  $15K > $10K!    │────────→│ tx-002: FRAUD│
│ tx-003: $120 │────────────→│  Analyze...      │         │              │
└─────────────┘             └──────────────────┘         └──────────────┘

2. Create a Fleet of Agents

Deploy 10 instances of the same agent

curl

1curl -X POST http://localhost:8080/api/v1/agents \
2  -H "Content-Type: application/json" \
3  -d '{
4    "name": "log-analyzer",
5    "engine": "rule-based",
6    "inputTopic": "application-logs",
7    "outputTopic": "log-anomalies",
8    "instances": 10,
9    "description": "Analyzes logs for anomalies (10 parallel instances)"
10  }'

Scale an existing agent

curl

1# Scale from 1 to 50 instances
2curl -X PUT http://localhost:8080/api/v1/agents/agent-a1b2c3/scale \
3  -H "Content-Type: application/json" \
4  -d '{"instances": 50}'

Bulk create multiple agents

curl

1curl -X POST http://localhost:8080/api/v1/agents/bulk \
2  -H "Content-Type: application/json" \
3  -d '[
4    {"name":"email-classifier",  "engine":"llm",        "inputTopic":"emails",            "outputTopic":"classified-emails","instances":5},
5    {"name":"spam-filter",       "engine":"rule-based",  "inputTopic":"classified-emails", "outputTopic":"clean-emails",     "instances":3},
6    {"name":"sentiment-analyzer","engine":"llm",        "inputTopic":"clean-emails",       "outputTopic":"email-insights",   "instances":2}
7  ]'

Programmatic fleet creation (Java)

Java

1import java.net.URI;
2import java.net.http.*;
3
4public class FleetExample {
5    public static void main(String[] args) throws Exception {
6        HttpClient http = HttpClient.newHttpClient();
7        String baseUrl = "http://localhost:8080/api/v1/agents";
8
9        String[] regions = {"us-east", "us-west", "eu-west", "eu-east", "ap-south"};
10
11        for (String region : regions) {
12            String json = """
13                {
14                  "name": "sensor-monitor-%s",
15                  "engine": "rule-based",
16                  "inputTopic": "sensors-%s",
17                  "outputTopic": "alerts-%s",
18                  "instances": 3,
19                  "description": "Monitors IoT sensors in %s"
20                }
21                """.formatted(region, region, region, region);
22
23            HttpRequest req = HttpRequest.newBuilder()
24                .uri(URI.create(baseUrl))
25                .header("Content-Type", "application/json")
26                .POST(HttpRequest.BodyPublishers.ofString(json))
27                .build();
28
29            HttpResponse<String> resp = http.send(req,
30                HttpResponse.BodyHandlers.ofString());
31            System.out.printf("Created sensor-monitor-%s: %s%n",
32                region, resp.body());
33        }
34        System.out.println("\nFleet deployed: 5 agents, 15 instances total");
35    }
36}

Output

1Created sensor-monitor-us-east: {"id":"agent-001","status":"RUNNING","instances":3}
2Created sensor-monitor-us-west: {"id":"agent-002","status":"RUNNING","instances":3}
3Created sensor-monitor-eu-west: {"id":"agent-003","status":"RUNNING","instances":3}
4Created sensor-monitor-eu-east: {"id":"agent-004","status":"RUNNING","instances":3}
5Created sensor-monitor-ap-south: {"id":"agent-005","status":"RUNNING","instances":3}
6
7Fleet deployed: 5 agents, 15 instances total

3. Create a Graph of Agents (Pipeline)

A graph connects multiple agents into a processing pipeline. Events flow from one agent to the next, with conditional routing and parallel processing.

Fluent DSL (Java)

Java

1import com.streamflow.graph.dsl.StreamflowGraph;
2import com.streamflow.graph.dsl.CompiledGraph;
3import com.streamflow.graph.dsl.GraphRunner;
4import java.util.Map;
5
6public class OrderProcessingPipeline {
7    public static void main(String[] args) throws Exception {
8
9        CompiledGraph graph = StreamflowGraph.define("order-processing")
10            .name("Order Processing Pipeline")
11            .version("1.0")
12
13            .start("validate")
14                .agent("order-validator")
15                .prompt("Check order for: valid product ID, positive quantity, valid address.")
16
17            .then("score-risk")
18                .agent("risk-scorer")
19                .prompt("Score order risk 0-100.")
20
21            .route("risk-route")
22                .when("state.riskScore < 50")
23                    .to("auto-approve")
24                .otherwise()
25                    .to("manual-review")
26
27            .end("auto-approve")
28                .agent("order-approver")
29                .prompt("Approve this order. Generate confirmation number.")
30
31            .end("manual-review")
32                .agent("review-flagger")
33                .prompt("Flag this order for manual review.")
34
35            .compile();
36
37        System.out.println("Graph compiled: " + graph.getNodes().size() + " nodes");
38
39        GraphRunner runner = new GraphRunner();
40        runner.register(graph);
41
42        Map<String, Object> order = Map.of(
43            "orderId", "ORD-2024-001",
44            "product", "Laptop Pro 16",
45            "quantity", 1,
46            "amount", 2499.99,
47            "customer", "Alice Martin",
48            "shippingCountry", "FR"
49        );
50
51        var result = runner.execute("order-processing", order);
52        System.out.printf("Result: %s%n", result.finalState());
53        System.out.printf("Path: %s%n", result.executedNodes());
54        System.out.printf("Duration: %d ms%n", result.durationMs());
55    }
56}

Output

1Graph compiled: 5 nodes
2Result: {status=APPROVED, confirmationId=CONF-7X2K9, riskScore=23}
3Path: [validate → score-risk → risk-route → auto-approve]
4Duration: 450 ms

Visual flow

         ┌─────────────┐     ┌─────────────┐     ┌──────────────┐
Order ──→│  validate    │────→│  score-risk  │────→│  risk-route  │
         │ (check data) │     │ (score 0-100)│     │  (if < 50)   │
         └─────────────┘     └─────────────┘     └──────┬───────┘
                                                        │
                              ┌──────────────────────────┤
                              ↓                          ↓
                    ┌─────────────────┐       ┌──────────────────┐
                    │  auto-approve   │       │  manual-review   │
                    │  (confirm order)│       │  (flag for human)│
                    └─────────────────┘       └──────────────────┘

Parallel Processing in Graphs

Java

1CompiledGraph graph = StreamflowGraph.define("content-moderation")
2    .start("extract")
3        .agent("content-extractor")
4        .prompt("Extract text, images, and links from the content")
5
6    .parallel("analyze")
7        .branch("text-check")
8            .agent("text-moderator")
9            .prompt("Check text for hate speech, spam, misinformation")
10        .branch("image-check")
11            .agent("image-moderator")
12            .prompt("Check images for NSFW, violence, copyright")
13        .branch("link-check")
14            .agent("link-checker")
15            .prompt("Verify all links are safe (no phishing, malware)")
16    .merge("combine-results")
17
18    .end("decide")
19        .agent("moderation-decision")
20        .prompt("Based on analysis, decide: APPROVE, REJECT, or FLAG")
21
22    .compile();

              ┌── text-moderator ──┐
extract ──┬──→│                    │──┬── combine ── decide
          │   └────────────────────┘  │
          ├──→ image-moderator ───────┤
          │                           │
          └──→ link-checker ──────────┘

4. Auto-Scaling

Configure via REST API

curl

1curl -X PUT http://localhost:8080/api/v1/agents/agent-a1b2c3/autoscale \
2  -H "Content-Type: application/json" \
3  -d '{
4    "enabled": true,
5    "minInstances": 2,
6    "maxInstances": 100,
7    "targetLatencyMs": 500,
8    "queueHighWatermark": 5000,
9    "queueLowWatermark": 100,
10    "cooldownSeconds": 60
11  }'

How auto-scaling works

Load increases                          Load decreases
     ↓                                       ↓
Queue > 5000                            Queue < 100
   OR                                     AND
Latency > 500ms                         Latency < 250ms
   OR                                     AND
Error rate > 5%                         Error rate < 2.5%
     ↓                                     AND
  SCALE UP                              Sustained 3× cooldown
  (immediate)                                ↓
                                         SCALE DOWN
                                         (conservative)

Asymmetry by design: Scale UP is aggressive (any single condition triggers). Scale DOWN is conservative (all conditions + stability period).

5. Monitor Agents in the UI

Open the StreamFlow Dashboard at http://localhost:8080.

How the UI reflects programmatic changes

Your code:  POST /api/v1/agents  →  Agent created
                                         ↓
UI:         /agents page         →  New card appears (real-time)
                                         ↓
Your code:  PUT /agents/{id}/scale  →  Instances: 1 → 50
                                         ↓
UI:         /agents/{id} page    →  Instance count updates

The UI polls the same REST API every 5 seconds. No separate registration needed.

API endpoints used by the UI

UI Action	API Call
List agents	GET /api/v1/agents
Agent detail	GET /api/v1/agents/{id}
Create agent	POST /api/v1/agents
Edit agent	PUT /api/v1/agents/{id}
Delete agent	DELETE /api/v1/agents/{id}
Pause	PUT /api/v1/agents/{id}/pause
Resume	PUT /api/v1/agents/{id}/resume
Scale	PUT /api/v1/agents/{id}/scale
Autoscale	PUT /api/v1/agents/{id}/autoscale

6. Agent Lifecycle Management

States

IDLE → EVENT_RECEIVED → CONTEXT_LOADING → REASONING → TOOL_CALLS → DECISION → ACTION
  ↑                                                                              │
  └──────────────────────────────────────────────────────────────────────────────┘

Control via REST

curl

1# Pause (stops processing, keeps state)
2curl -X PUT http://localhost:8080/api/v1/agents/agent-a1b2c3/pause
3
4# Resume (continues from last offset)
5curl -X PUT http://localhost:8080/api/v1/agents/agent-a1b2c3/resume
6
7# Restart (drains queue, reloads config)
8curl -X PUT http://localhost:8080/api/v1/agents/agent-a1b2c3/restart
9
10# Delete (stops and removes)
11curl -X DELETE http://localhost:8080/api/v1/agents/agent-a1b2c3

Control via Java

Java

1HttpClient http = HttpClient.newHttpClient();
2String agentUrl = "http://localhost:8080/api/v1/agents/agent-a1b2c3";
3
4// Pause
5http.send(HttpRequest.newBuilder()
6    .uri(URI.create(agentUrl + "/pause"))
7    .PUT(HttpRequest.BodyPublishers.noBody()).build(),
8    HttpResponse.BodyHandlers.ofString());
9
10// Scale to 20 instances
11http.send(HttpRequest.newBuilder()
12    .uri(URI.create(agentUrl + "/scale"))
13    .header("Content-Type", "application/json")
14    .PUT(HttpRequest.BodyPublishers.ofString("{\"instances\": 20}")).build(),
15    HttpResponse.BodyHandlers.ofString());

7. Agent Failover (High Availability)

Primary/Standby Architecture

Region A (primary)                Region B (standby)
┌──────────────────┐             ┌──────────────────┐
│ fraud-detector   │  heartbeat  │ fraud-detector   │
│ Status: ACTIVE   │────────────→│ Status: STANDBY  │
│ Offset: 42,000   │  (every 5s) │ Offset: 42,000   │
└──────────────────┘             └──────────────────┘
         ↓                                ↓
   Region A dies                   Heartbeat timeout (15s)
                                         ↓
                                  FENCING (5s window)
                                         ↓
                                  PROMOTE to ACTIVE
                                  Resume from offset 42,000
                                         ↓
                              ┌──────────────────┐
                              │ fraud-detector   │
                              │ Status: ACTIVE   │
                              │ Offset: 42,000   │ ← zero duplicate processing
                              └──────────────────┘

Guarantees

Zero re-processing: Standby resumes from the exact checkpoint offset
Split-brain prevention: Fencing token ensures only one replica can promote
State replication: Agent state replicated via Raft (RF=3)
Automatic: No manual intervention needed

Summary

What	How
Create 1 agent	POST /api/v1/agents
Create fleet	POST /api/v1/agents/bulk
Create graph	StreamflowGraph.define()
Scale agent	PUT /agents/{id}/scale
Auto-scale	PUT /agents/{id}/autoscale
Monitor	UI at :8080
Pause/Resume	PUT /agents/{id}/pause\|resume
Failover	Automatic (heartbeat + fencing)

StreamFlow Agent Guide

Table of Contents

1. Create a Single Agent

Via REST API

Response

Via Java SDK (Programmatic)

Output

How it works

2. Create a Fleet of Agents

Deploy 10 instances of the same agent

Scale an existing agent

Bulk create multiple agents

Programmatic fleet creation (Java)

Output

3. Create a Graph of Agents (Pipeline)

Fluent DSL (Java)

Output

Visual flow

Parallel Processing in Graphs

4. Auto-Scaling

Configure via REST API

How auto-scaling works

5. Monitor Agents in the UI

Agents List Page (/agents)

Agent Detail Page (/agents/{id})

How the UI reflects programmatic changes

API endpoints used by the UI

6. Agent Lifecycle Management

States

Control via REST

Control via Java

7. Agent Failover (High Availability)

Primary/Standby Architecture

Guarantees

Summary