Skip to main content

​
Agent Builder SDK

The Agent Builder SDK lets you build your own AI agents — voice bots, chat assistants, automation workflows, IoT controllers, or any custom agent — and deploy them on the RBS platform.
This is separate from the RBS coding agent. The coding agent (rbs agent) is a built-in AI coding assistant. The Agent Builder SDK (rbs agent-builder) is an SDK for you to create and deploy your own agents for any purpose.

​
Overview

Event-driven

Process typed events through a message bus. Route, filter, transform, and broadcast events between components.

Node-based

Build agents from composable nodes — reasoning (LLM), tool execution, routing, and custom logic.

DAG workflows

Orchestrate multi-step workflows with dependency-based execution and parallel processing.

RBS DSL configuration

Define everything in BUILD.rbs files. No boilerplate, no YAML, no separate config files.

​
Quick start

​
1. Define an agent node

# BUILD.rbs

# Define a customer support reasoning node
native.define_agent_node(
    name = "customer_support",
    description = "Customer support agent using LLM",
    type = "reasoning",
    system_prompt = """You are a helpful customer support agent.
    Be friendly, professional, and solve customer issues efficiently.""",
    model = "gpt-4",
    tools = ["lookup_order", "create_ticket", "transfer_agent"],
)

​
2. Define a bridge for event routing

# Define routing for the support node
native.define_agent_bridge(
    name = "support_bridge",
    node = "customer_support",
    patterns = ["user.*", "agent.*"],
)

​
3. Define the agent system

# Define the complete agent system
native.define_agent_system(
    name = "support_agent",
    description = "Customer support voice/chat agent",
    harness = {
        "type": "websocket",
        "port": 8080,
        "path": "/agent",
    },
    nodes = ["customer_support"],
    bridges = ["support_bridge"],
    speaking_node = "customer_support",
)

​
4. Run the agent

# Start the agent
rbs agent-builder serve //path/to:support_agent

# Test the agent
rbs agent-builder test //path/to:support_agent \
    --event "user.message" \
    --data '{"content": "I need help with my order"}'

​
Core concepts

​
Events

Events are typed messages that flow through the system. Every user input, agent response, tool call, and system notification is an event.
Event typeDescription
user.messageUser text input.
user.started_speakingUser began speaking (voice).
user.stopped_speakingUser stopped speaking.
agent.responseAgent text response.
agent.speech_sentAgent audio sent.
tool.callTool invocation.
tool.resultTool result.
session.startSession began.
session.endSession ended.
system.errorAn error occurred.

​
Nodes

Nodes are the building blocks of agents. Each node processes events and produces new events.
Node typeDescription
ReasoningLLM-powered conversation handling. Receives user messages, generates responses, calls tools.
ToolExecutes a specific action (API call, database query, file operation).
RouterRoutes events to other nodes based on conditions (intent classification, priority routing).
CustomAny custom logic defined in an implementation function.

​
Bridges

Bridges connect nodes to the event bus. They handle:
  • Event pattern matching — Subscribe to specific event types (e.g., user.*).
  • Filtering — Drop events that don’t match criteria.
  • Transformation — Transform events before passing to the node.
  • Broadcasting — Send events from the node back to the bus.

​
Architecture

┌──────────────────────────────────────────────────────┐
│                    AgentSystem                         │
│  ┌──────────────────────────────────────────────┐    │
│  │                    Bus                        │    │
│  │  (Event routing, message queuing)             │    │
│  └──────────────────────┬───────────────────────┘    │
│                          │                            │
│  ┌──────────┐   ┌──────────┐   ┌──────────┐         │
│  │ Bridge 1 │   │ Bridge 2 │   │ Bridge N │         │
│  └─────┬────┘   └─────┬────┘   └─────┬────┘         │
│        │               │              │               │
│  ┌─────┴────┐   ┌─────┴────┐   ┌─────┴────┐         │
│  │  Node 1  │   │  Node 2  │   │  Node N  │         │
│  │(Reasoning)│   │  (Tool)  │   │ (Router) │         │
│  └──────────┘   └──────────┘   └──────────┘         │
│                                                       │
│  ┌──────────────────────────────────────────────┐    │
│  │                  Harness                      │    │
│  │  (WebSocket/HTTP I/O, input/output queues)    │    │
│  └──────────────────────────────────────────────┘    │
└──────────────────────────────────────────────────────┘

​
Defining nodes

​
Reasoning node (LLM-powered)

native.define_agent_node(
    name = "assistant",
    description = "General assistant with tool access",
    type = "reasoning",
    system_prompt = """You are a helpful assistant that can look up orders,
    check inventory, and process returns.""",
    model = "gpt-4",
    tools = ["lookup_order", "check_inventory", "process_return"],
    config = {
        "temperature": 0.7,
        "max_tokens": 1024,
    },
)

​
Tool node

native.define_agent_node(
    name = "order_lookup",
    description = "Look up order details from the database",
    type = "tool",
    implementation = _lookup_order_impl,
    inputs = {
        "order_id": attr.string(mandatory = True),
    },
    outputs = {
        "status": attr.string(),
        "items": attr.list(),
        "total": attr.float(),
    },
)

​
Router node

native.define_agent_node(
    name = "intent_router",
    description = "Route user messages to the correct handler",
    type = "router",
    implementation = _intent_router_impl,
    config = {
        "routes": {
            "order_inquiry": "order_handler",
            "billing_question": "billing_handler",
            "technical_issue": "tech_support",
            "default": "general_assistant",
        },
    },
)

​
Custom node

def _custom_node_impl(ctx):
    event = ctx.event
    if event.type == "user.message":
        # Custom processing logic
        enriched = enrich_message(event.content)
        return ctx.emit("agent.response", content = enriched)

native.define_agent_node(
    name = "custom_processor",
    description = "Custom event processing node",
    type = "custom",
    implementation = _custom_node_impl,
)

​
Defining bridges

native.define_agent_bridge(
    name = "support_bridge",
    description = "Routes user events to the support node",
    node = "customer_support",
    patterns = ["user.*"],                 # Subscribe to all user events
    routes = [
        native.route(
            on = "user.message",
            action = "map",
            handler = _process_message,    # Transform before passing to node
            filter = _is_english,          # Only process English messages
        ),
        native.route(
            on = "user.started_speaking",
            action = "broadcast",          # Broadcast to all nodes
        ),
    ],
    auth = {
        "required": True,
        "roles": ["customer", "admin"],
    },
)

​
DAG workflows

Define complex multi-step workflows with dependencies: 
native.define_agent_dag(
    name = "order_processing",
    description = "Process customer orders end-to-end",

    nodes = {
        "validate": native.node_ref("validator"),
        "authenticate": native.node_ref("auth_node", deps = ["validate"]),
        "process": native.node_ref("processor", deps = ["authenticate"]),
        "confirm": native.node_ref("confirmation", deps = ["process"]),
    },

    entry_nodes = ["validate"],
    exit_nodes = ["confirm"],
    on_error = "error_handler",
    timeout_seconds = 300,
    max_parallel_nodes = 5,
)

​
DAG execution flow

[validate] → [authenticate] → [process] → [confirm]
Each node:
  • Receives input from completed upstream nodes.
  • Runs independently with its own context.
  • Passes results downstream.
  • Supports conditional execution and looping.

​
Node references

native.node_ref(
    node = "processor",
    deps = ["authenticate"],       # Must complete before this node starts
    condition = lambda ctx: ctx.inputs.prev.value > 0,  # Conditional execution
    wait_for = "all",              # "all" (all deps) or "any" (first dep)
    loop_to = "validate",          # Loop back to a previous node
    max_iterations = 3,            # Maximum loop iterations
)

​
Agent systems

Combine nodes, bridges, and a harness into a complete agent system: 
native.define_agent_system(
    name = "support_system",
    description = "Customer support agent with voice and chat",
    harness = {
        "type": "websocket",
        "port": 8080,
        "path": "/agent",
    },
    nodes = [
        "intent_router",
        "order_handler",
        "billing_handler",
        "general_assistant",
    ],
    bridges = [
        "user_bridge",
        "order_bridge",
        "billing_bridge",
        "assistant_bridge",
    ],
    speaking_node = "general_assistant",
    middleware = ["auth_middleware", "logging_middleware"],
    config = {
        "max_concurrent_sessions": 100,
        "session_timeout": 3600,
    },
)

​
CLI commands

​
Serve

Start an agent system: 
# Start the agent
rbs agent-builder serve //path/to:support_system

# Custom port
rbs agent-builder serve //path/to:support_system --port 9000

# Hot reload on file changes
rbs agent-builder serve //path/to:support_system --watch

​
Test

Test an agent with events: 
# Send a single event
rbs agent-builder test //path/to:support_system \
    --event "user.message" \
    --data '{"content": "What is the status of order #12345?"}'

# Interactive testing mode
rbs agent-builder test //path/to:support_system --interactive

​
List

List registered components: 
rbs agent-builder list
rbs agent-builder list --verbose

​
DAG commands

# Show DAG structure
rbs agent-builder dag show order_workflow

# Validate DAG (check for cycles, missing deps)
rbs agent-builder dag validate order_workflow

# Export as Mermaid or DOT diagram
rbs agent-builder dag export order_workflow --format mermaid
rbs agent-builder dag export order_workflow --format dot

​
RBS DSL API reference

​
native.define_agent_node()

native.define_agent_node(
    name = "my_node",               # Required: unique name
    description = "...",            # Optional: description
    type = "reasoning",             # "reasoning", "tool", "router", "custom"
    system_prompt = "...",          # LLM system prompt (reasoning nodes)
    model = "gpt-4",               # LLM model (reasoning nodes)
    tools = ["tool1", "tool2"],     # Available tools (reasoning nodes)
    implementation = func,          # Custom implementation function
    inputs = {...},                 # Input schema (tool/custom nodes)
    outputs = {...},                # Output schema (tool/custom nodes)
    config = {...},                 # Additional configuration
)

​
native.define_agent_bridge()

native.define_agent_bridge(
    name = "my_bridge",             # Required: unique name
    description = "...",            # Optional: description
    node = "node_name",             # Associated node
    routes = [...],                 # Route definitions
    patterns = ["user.*"],          # Event patterns to subscribe to
    auth = {...},                   # Authorization configuration
)

​
native.define_agent_system()

native.define_agent_system(
    name = "my_system",             # Required: unique name
    description = "...",            # Optional: description
    harness = {...},                # I/O configuration (websocket, http)
    nodes = ["node1"],              # Node names
    bridges = ["bridge1"],          # Bridge names
    speaking_node = "node1",        # Initial speaking node
    middleware = [...],             # Middleware chain
    config = {...},                 # Additional configuration
)

​
native.define_agent_dag()

native.define_agent_dag(
    name = "my_workflow",           # Required: unique name
    description = "...",            # Optional: description
    nodes = {...},                  # Node references dict
    entry_nodes = [...],            # Entry points
    exit_nodes = [...],             # Exit points
    on_error = "handler",           # Error handler node
    timeout_seconds = 300,          # Execution timeout
    max_parallel_nodes = 5,         # Parallelism limit
    config = {...},                 # Additional configuration
)

​
Helper functions

# Node reference for DAGs
native.node_ref(
    node = "node_name",
    deps = ["dep1", "dep2"],
    condition = lambda ctx: ctx.inputs.prev.value > 0,
    wait_for = "all",
    loop_to = "other_node",
    max_iterations = 5,
)

# Route definition for bridges
native.route(
    on = "user.message",
    action = "map",                 # "map", "filter", "broadcast", "stream"
    handler = my_handler,
    filter = my_filter,
    dag = "workflow_name",
)

# Input reference for DAG data flow
native.input_from(
    node = "upstream_node",
    field = "output_field",
)

# Output schema definition
native.output_schema(
    schema = {
        "result": attr.string(),
        "score": attr.float(),
    },
)

​
Example: Multi-node support agent

# BUILD.rbs — Complete customer support agent

# Tools
def _lookup_order_impl(ctx):
    order_id = ctx.attr.order_id
    # Query database for order details
    return ctx.output.success(data = {"order_id": order_id, "status": "shipped"})

native.define_agent_node(
    name = "order_lookup_tool",
    type = "tool",
    description = "Look up order details",
    implementation = _lookup_order_impl,
    inputs = {"order_id": attr.string(mandatory = True)},
)

# Reasoning nodes
native.define_agent_node(
    name = "support_agent",
    type = "reasoning",
    description = "Main customer support agent",
    system_prompt = """You are a customer support agent. Be helpful and professional.
    Use the order_lookup tool to check order status when asked.""",
    model = "gpt-4",
    tools = ["order_lookup_tool"],
)

native.define_agent_node(
    name = "escalation_agent",
    type = "reasoning",
    description = "Handles escalated issues",
    system_prompt = "You handle complex escalated customer issues.",
    model = "gpt-4",
    tools = ["order_lookup_tool", "create_ticket"],
)

# Router
def _route_by_sentiment(ctx):
    if "angry" in ctx.event.content.lower() or "frustrated" in ctx.event.content.lower():
        return "escalation_agent"
    return "support_agent"

native.define_agent_node(
    name = "sentiment_router",
    type = "router",
    description = "Routes by customer sentiment",
    implementation = _route_by_sentiment,
)

# Bridges
native.define_agent_bridge(
    name = "user_bridge",
    node = "sentiment_router",
    patterns = ["user.*"],
)

native.define_agent_bridge(
    name = "support_bridge",
    node = "support_agent",
    patterns = ["agent.*"],
)

native.define_agent_bridge(
    name = "escalation_bridge",
    node = "escalation_agent",
    patterns = ["agent.*"],
)

# System
native.define_agent_system(
    name = "customer_support",
    description = "Multi-node customer support system",
    harness = {
        "type": "websocket",
        "port": 8080,
        "path": "/support",
    },
    nodes = ["sentiment_router", "support_agent", "escalation_agent", "order_lookup_tool"],
    bridges = ["user_bridge", "support_bridge", "escalation_bridge"],
    speaking_node = "support_agent",
)
Run it: 
rbs agent-builder serve //support:customer_support
# Agent running at ws://localhost:8080/support

​
Coding agent vs Agent Builder SDK

Coding agentAgent Builder SDK
Commandrbs agentrbs agent-builder
PurposeAI coding assistant (like Cursor).SDK for building your own agents.
UsersDevelopers using RBS.Developers building agents.
ConfigurationEnvironment variables and rules files.RBS DSL in BUILD.rbs.
CustomizationCustom tools, prompts, and roles.Full agent architecture.
The only shared component is the tool system — you can use native.define_agent_tool() to create tools that work in both systems.