railtownai/railtracks/agent-builder

Build a Railtracks Agent

The user wants to build an agent using the railtracks framework: $ARGUMENTS

How railtracks works

• Tools are plain Python functions decorated with @rt.function_node. Type hints become the parameter schema; the docstring becomes the description.
• Agents are created with rt.agent_node(). Its behaviour depends on whether tools or a structured output schema are passed (see below).
• Flows wrap an agent or async function as the entry point and handle execution, config, and context.
• `rt.call()` is used inside async workflows to call agents or nodes directly.
• Results: flow.invoke(), await flow.ainvoke() and await rt.call() on an agent return a response object. Read it with .content: a str for a text agent, an instance of your schema for an output_schema agent. The full conversation is on .message_history. When the agent was given an output_schema, .structured also returns that instance; otherwise .text also returns the str.

What agent_node builds

rt.agent_node() builds one node behind the scenes — there is no separate named type to pick. What you pass changes what the agent does at runtime:

LLM Providers

# Example model IDs; use the provider and model the project already uses
rt.llm.AnthropicLLM("claude-sonnet-5-5")
rt.llm.OpenAILLM("gpt-6-luna")
rt.llm.GeminiLLM("gemini-3.8-flash")
rt.llm.OpenAICompatibleProvider(
    "my-model", api_base="https://api.example.com/v1", api_key="..."
)

Steps

1. Read the existing code — check what files already exist in the project. Understand the task before writing anything.
2. Identify what tools the agent needs — each capability the agent should have becomes a @rt.function_node. Ask the user to clarify if it's not obvious from $ARGUMENTS.
3. Define the tools — write each tool as a Python function with:

• Full type hints on all parameters and return value
• A docstring with a one-line summary and Args: / Returns: sections
• Real implementation (or a clear stub with a TODO if the user needs to fill it in)

1. Define the agent — call rt.agent_node() with (note: it returns a class/type, so use PascalCase for the variable name):

• A descriptive name
• tool_nodes listing the tools (if any), or output_schema as a Pydantic BaseModel for structured output — one or the other, never both (passing both raises NodeCreationError)
• llm — follow the provider the project already uses, or one whose API key is configured (providers and their key variables: https://docs.railtracks.org/documentation/getting_started/llm_setup/); otherwise ask the user which LLM to use
• system_message — a clear, specific system prompt

1. Wrap in a Flow — create rt.Flow(name="...", entry_point=MyAgent) for simple cases. For multi-step or multi-agent workflows, define an async def function as the entry point and use await rt.call(MyAgent, ...) inside it.
2. Add invocation code — include a if __name__ == "__main__": block that calls flow.invoke(...) with a representative example and prints result.content, so the user can run it immediately. Use await flow.ainvoke(...) instead if the code is already async.
3. Check imports — make sure import railtracks as rt is at the top and any Pydantic models import from pydantic import BaseModel.

Patterns to Follow

Simple Agent with Tools

import railtracks as rt
@rt.function_node
def my_tool(param: str) -> str:
    """One-line description.
    Args:
        param: What this parameter is.
    Returns:
        What this returns.
    """
    return f"result for {param}"
llm = rt.llm.AnthropicLLM("claude-sonnet-5-5")
# agent_node returns a class (type), not an instance — use PascalCase
MyAgent = rt.agent_node(
    "Agent Name",
    tool_nodes=[my_tool],
    llm=llm,
    system_message="You are a helpful assistant that ...",
)
flow = rt.Flow(name="My Flow", entry_point=MyAgent)
if __name__ == "__main__":
    result = flow.invoke("user query here")
    print(result.content)

Structured Output

from pydantic import BaseModel
class Output(BaseModel):
    field1: str
    field2: int
StructuredAgent = rt.agent_node(
    "Structured Agent",
    output_schema=Output,
    llm=llm,
)
flow = rt.Flow(name="Structured Flow", entry_point=StructuredAgent)
result = flow.invoke("user query here")
print(result.content.field1)  # .content is an Output instance

Decision Node (Classify, Route, Grade)

When the output is a judgment you act on in code (yes/no, one of a fixed set of labels, a level on a rubric), use rt.decision_node with a decision model instead of an LLM. One request answers every question with calibrated probabilities. Input is text, or a JSON object or array; no images.

is_urgent = rt.decisions.Predicate(name="is_urgent", instructions="The message is urgent")
team = rt.decisions.Choice(
    name="team", instructions="Which team should handle this", choices=["billing", "technical"]
)
schema = rt.decisions.DecisionSchema(predicate=[is_urgent], choice=[team])
# or OpenRouterDecisions("typesafe/jev-1.13") / OpenAIDecisions("gpt-6-luna")
Triage = rt.decision_node(
    "Triage", model=rt.decisions.TypeSafeDecisions("jev-latest"), schema=schema
)
result = rt.Flow(name="Triage", entry_point=Triage).invoke("Checkout is down!")
if result.structured[is_urgent].probability >= 0.8:  # typed PredicateAnswer
    print(result.structured[team].choice)

It's also a tool: pass it in an agent's tool_nodes. Failures raise rt.exceptions.DecisionModelError. Only node middleware (rt.wrap_node, rt.post_node) works; model middleware such as pre_llm raises NodeCreationError. Docs: https://docs.railtracks.org/documentation/agent_design/decision_nodes/

Multi-Agent Workflow

@rt.function_node
async def pipeline(query: str) -> str:
    step1 = await rt.call(AgentA, query)
    step2 = await rt.call(AgentB, step1.content)
    return step2.content
flow = rt.Flow(name="Pipeline", entry_point=pipeline)

Async Invocation

import asyncio
async def main():
    result = await flow.ainvoke("user query here")
    print(result.content)
if __name__ == "__main__":
    asyncio.run(main())

Use flow.invoke() from sync code and await flow.ainvoke() from async code.

Streaming

rt.astream streams an agent's text as it's generated. It's async only; there is no stream=True on models.

async def main():
    stream = rt.astream(MyAgent, user_input="user query here")
    async for chunk in stream:
        print(chunk, end="", flush=True)  # str chunks
    final = stream.result  # the complete response; read final.content

await rt.astream(...) without the loop returns just the final response.

Agent Used as a Tool by Another Agent (Multi-Agent Orchestration)

An agent with a system_message can be passed straight into another agent's tool_nodes. Without a manifest=, railtracks names the tool after the agent (characters outside [a-zA-Z0-9_-] become _), builds the description from the agent's name and full system message, and gives it one required string parameter, request, which the sub-agent receives as its user message. An agent with neither system_message nor manifest= raises NodeCreationError when used as a tool. {placeholders} in the system message are never filled in the description, because ContextInjection fills messages only; pass a manifest when the sub-agent's system message is templated.

SubAgent = rt.agent_node(
    "Sub Agent",
    tool_nodes=[tool_a],
    llm=llm,
    system_message="You do X for a given topic.",
)
Orchestrator = rt.agent_node(
    "Orchestrator",
    tool_nodes=[SubAgent],  # the orchestrator calls Sub_Agent(request=...)
    llm=llm,
    system_message="You are an orchestrator. Delegate to sub-agents as needed.",
)

Pass rt.ToolManifest to agent_node when the caller needs a shorter description than the system message, named parameters, or a description that does not reveal the prompt. A manifest without parameters keeps its description and takes request. A manifest with a blank description raises NodeCreationError.

from railtracks.llm import Parameter
SubAgent = rt.agent_node(
    "Sub Agent",
    tool_nodes=[tool_a],
    llm=llm,
    manifest=rt.ToolManifest(
        description="Does X given a topic. Call this when you need X.",
        parameters=[
            Parameter(
                name="topic", description="The topic to process", param_type="string"
            ),
        ],
    ),
)
Orchestrator = rt.agent_node(
    "Orchestrator",
    tool_nodes=[SubAgent],  # SubAgent is now a tool the orchestrator can call
    llm=llm,
    system_message="You are an orchestrator. Delegate to sub-agents as needed.",
)

Parameter fields:

• name — the argument name the orchestrator LLM passes
• description — explains what to put in this argument
• param_type — JSON schema type string ("string", "integer", "number", "boolean", …) or a Python builtin mapped the same way: str, int, float (→ "number"), bool, list / tuple / set (→ "array"), dict (→ "object"), type(None) (→ "null"). Unknown types fall back to "object".
• required — defaults to True
• enum — optional list of allowed values

MCP Tools

server = rt.connect_mcp(
    rt.MCPStdioParams(command="python", args=["-m", "my_mcp_server"])
)
MCPAgent = rt.agent_node("MCP Agent", tool_nodes=server.tools, llm=llm)

Visualizing Runs

Every run is recorded to .railtracks/ automatically, no code needed. To inspect runs in the local visualizer, have the user run these from the project root:

pip install 'railtracks[visual]'
railtracks init
railtracks viz --beta

railtracks viz --beta downloads the beta UI on first use and serves it at http://localhost:3031. It blocks, so don't start it from inside the agent script.

Rules

• Pass llm= to every rt.agent_node(...); it's required.
• Add guardrails with @rt.input_guard / @rt.output_guard in model_middleware=[...].
• Import prebuilt guards (BlockTextInputGuard, PIIRedactConfig, ScopeGuard, …) from railtracks.prebuilt.guardrails.
• Stream with rt.astream(...).
• Run agents through rt.Flow(...).
• Give each agent one entry point: a Flow, or await rt.call() inside another node.
• Write each tool's docstring as a clear, specific description; it's the tool description the LLM sees.
• Type-hint every tool parameter and return value; the hints define the tool's parameter schema.
• Give each agent only the tools it needs.