tantra

package module
v0.0.0-...-aecfaea Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Feb 7, 2026 License: MIT Imports: 23 Imported by: 0

README

tantra-go

Go Reference CI Go Report Card

Minimal Go framework for LLM agents. Single agents, tool calling, skills, multi-agent swarms, graph workflows, multimodal input, streaming, MCP integration, checkpointing, observability, HTTP serving.

go get github.com/tantra-run/tantra-go

Requires Go 1.22+.

Why tantra-go

Most LLM frameworks are either Python-only or bury simple concepts under layers of abstraction. tantra-go is built on a few observations:

  • An agent is just a loop. Send messages to an LLM, execute tool calls, repeat. That's Agent.Run — about 40 lines of actual logic. Everything else is composition on top.
  • Multi-agent is a routing problem, not a framework problem. Swarm lets the LLM decide routing at runtime. Graph lets you decide routing at build time. Pick the one that matches your problem.
  • State is the hard part. When agents hand off to each other, structured data needs to flow without serialization loss. A single shared *State with scoped access solves this without message-passing gymnastics.
  • You shouldn't need a framework to switch providers. Completer is one method. Implement it for Anthropic, Gemini, local models, or a mock for tests. The OpenAI provider is included because it covers most use cases (and works with any OpenAI-compatible API via WithBaseURL).

Quick start

package main

import (
    "context"
    "fmt"

    "github.com/tantra-run/tantra-go"
)

func main() {
    agent := &tantra.Agent{
        Name:         "assistant",
        Completer:    tantra.NewOpenAI(), // reads OPENAI_API_KEY from env
        SystemPrompt: "You are a helpful assistant.",
    }

    result, err := agent.Run(context.Background(), "What is Go?")
    if err != nil {
        panic(err)
    }
    fmt.Println(result.Output)
}

Agent

Agent wraps an LLM provider and optional tools. It loops: send messages to the LLM, execute any tool calls, repeat until the LLM responds with text (no tool calls) or MaxIterations is hit.

MaxIterations exists because LLMs can get stuck in tool-calling loops (tool A produces output that makes the LLM call tool A again). The default of 10 is generous for most use cases; lower it for tight cost control, raise it for complex research tasks.

flowchart LR
    A[User Input] --> B[LLM]
    B -->|tool calls| C[Execute Tools]
    C --> B
    B -->|text response| D[Result]

agent := &tantra.Agent{
    Name:          "researcher",
    Completer:     tantra.NewOpenAI(tantra.WithModel("gpt-4o")),
    Tools:         []tantra.Tool{searchTool, fetchTool},
    SystemPrompt:  "You are a research assistant.",
    MaxIterations: 5, // default: 10
}

result, err := agent.Run(ctx, "Find recent papers on RLHF")
// result.Output       - final text response
// result.Messages     - full conversation history (for multi-turn)
// result.ToolCalls    - total tool invocations
// result.Iterations   - LLM round-trips
// result.PromptTokens / result.OutputTokens
// result.DurationMS
// result.Cost         - if provider implements CostEstimator
Run options
// Share state across runs or agents
state := tantra.NewState()
result, _ := agent.Run(ctx, "hello", tantra.WithRunState(state))

// Multimodal input (see Multimodal section)
result, _ := agent.Run(ctx, "", tantra.WithParts(
    tantra.TextPart("What's in this image?"),
    tantra.ImageURLPart("https://example.com/photo.jpg"),
))

// Multi-turn conversation (see Multi-turn section)
result, _ := agent.Run(ctx, "Follow-up question", tantra.WithMessages(history))

// Observe internal events (see Observability section)
result, _ := agent.Run(ctx, "hello", tantra.WithObserver(myObserver))

Multi-turn

By default, Agent.Run starts a fresh conversation each time — system prompt plus one user message. For chat applications or multi-step reasoning, pass conversation history with WithMessages. The agent prepends its system prompt (if not already in the history), appends your new input as a user message, runs the loop, and returns the full message history in Result.Messages so you can feed it back on the next turn.

agent := &tantra.Agent{
    Name:         "chat",
    Completer:    tantra.NewOpenAI(),
    SystemPrompt: "You are a helpful assistant.",
}

// Turn 1
result, _ := agent.Run(ctx, "What is Go?")
// result.Messages = [system, user("What is Go?"), assistant("Go is...")]

// Turn 2 — pass previous messages back
result, _ = agent.Run(ctx, "Who created it?", tantra.WithMessages(result.Messages))
// result.Messages = [system, user("What is Go?"), assistant("Go is..."), user("Who created it?"), assistant("Rob Pike...")]

// Turn 3
result, _ = agent.Run(ctx, "When was it released?", tantra.WithMessages(result.Messages))

If the history already starts with a system message, the agent's SystemPrompt is not duplicated — the existing one is preserved. This lets you override the system prompt per-conversation if needed.

Result.Messages is always populated, even for single-turn calls. This means you can start simple and add multi-turn later without changing your code structure — just start passing result.Messages back.

Tools

Two ways to define tools. NewTool gives you compile-time type safety — the LLM's JSON arguments are unmarshaled into your struct, so you catch schema mismatches early and avoid map[string]any type assertions. SimpleTool is there for when the schema is dynamic or you just want something quick without defining a struct.

NewTool (type-safe, generics)

Define args as a struct with json tags. The schema is derived from struct tags automatically — no need to keep a separate JSON schema definition in sync with your Go types.

type WeatherArgs struct {
    City    string `json:"city" desc:"City name" required:"true"`
    Country string `json:"country" desc:"ISO country code"`
}

weatherTool := tantra.NewTool("get_weather", "Get current weather",
    func(ctx context.Context, state *tantra.ScopedState, args WeatherArgs) (tantra.ToolResult, error) {
        // args.City and args.Country are typed and populated
        weather := fetchWeather(args.City, args.Country)
        return tantra.SimpleResult(weather), nil
    })

Supported struct tag annotations:

  • json:"field_name" - parameter name (required)
  • desc:"description" - parameter description
  • required:"true" - marks parameter as required

Supported field types: string, int, float64, bool (mapped to JSON schema types string, integer, number, boolean).

SimpleTool (no generics)

Define params explicitly. Args arrive as map[string]any. Useful for tools whose parameters are determined at runtime (e.g., generated from a database schema) or when you don't want to define a struct for a one-off tool.

echoTool := tantra.SimpleTool("echo", "Echo back input",
    []tantra.Param{
        {Name: "text", Type: "string", Description: "Text to echo", Required: true},
    },
    func(ctx context.Context, state *tantra.ScopedState, args map[string]any) (tantra.ToolResult, error) {
        text, _ := args["text"].(string)
        return tantra.SimpleResult(text), nil
    })
Tool chaining

Sometimes a tool's output should deterministically feed into another tool without asking the LLM what to do next. Calling the LLM just to have it say "now call process with this data" wastes tokens and adds latency. ChainResult lets you short-circuit that: name the next tool and its arguments, and the agent executes the chain locally before returning to the LLM.

validate := tantra.SimpleTool("validate", "Validate input", nil,
    func(ctx context.Context, state *tantra.ScopedState, args map[string]any) (tantra.ToolResult, error) {
        data, _ := args["data"].(string)
        if data == "" {
            return tantra.SimpleResult("Invalid: empty data"), nil
        }
        // Chain to process tool — no LLM call in between
        return tantra.ChainResult("Valid", "process", map[string]any{"data": data}), nil
    })

process := tantra.SimpleTool("process", "Process data", nil,
    func(ctx context.Context, state *tantra.ScopedState, args map[string]any) (tantra.ToolResult, error) {
        return tantra.SimpleResult("Processed: " + args["data"].(string)), nil
    })

// Agent executes: validate -> process -> returns "Processed: ..." to LLM
agent := &tantra.Agent{
    Completer: provider,
    Tools:     []tantra.Tool{validate, process},
}

Chains are capped at 10 steps to prevent infinite loops. Only the final output is sent back to the LLM — intermediate outputs are tracked in Result.ToolResults for debugging.

flowchart LR
    LLM -->|calls| V[validate]
    V -->|ChainResult| P[process]
    P -->|final output| LLM
    style V fill:#f0f0f0,stroke:#333
    style P fill:#f0f0f0,stroke:#333
Tool interface

Both NewTool and SimpleTool implement:

type Tool interface {
    Name() string
    Description() string
    Schema() map[string]any
    Execute(ctx context.Context, state *ScopedState, args map[string]any) (ToolResult, error)
}

Implement this directly for full control over schema generation or execution.

Why tools receive ScopedState, not return state

Tools write state via state.Set() rather than returning state changes in ToolResult. This is deliberate: a tool might write multiple keys, conditionally write based on what it reads, or promote data to session scope — patterns that are awkward to express as a return value. ToolResult stays focused on what the LLM needs to see (output text) and what the agent loop needs (optional chain target).

Built-in tools

Agents that work with code need file and shell access. Rather than reimplementing the same primitives, tantra provides them as built-in tools.

agent := &tantra.Agent{
    Completer: provider,
    Tools:     tantra.CodeTools(), // read, write, edit, bash, think
}
Tool Description
ReadTool() Read a text file
WriteTool() Create or overwrite a file (creates parent dirs)
EditTool() Find and replace text in a file (must be unique match)
BashTool() Execute a shell command (inherits context, 64KB output cap)
ThinkTool() Reason step by step (echoes thought back into context)
CodeTools() All five as []Tool

All tools return errors as output strings (not Go errors) so the LLM can see what went wrong and self-correct. The edit tool requires old_string to appear exactly once — ambiguous edits are rejected. The bash tool uses exec.CommandContext to respect context cancellation and deadlines.

// Or pick individual tools
agent.Tools = []tantra.Tool{tantra.ReadTool(), tantra.BashTool()}
Autonomous agents

CodeTools() isn't just "file access for coding agents" — it's a self-sufficient foundation for autonomous agents. An agent with read, write, edit, bash, and think can create any capability it needs at runtime by writing scripts to disk and executing them. The filesystem becomes the tool registry.

agent := &tantra.Agent{
    Completer: provider,
    Tools:     tantra.CodeTools(),
    SystemPrompt: `You are an autonomous agent. To solve problems:
1. Use think to plan your approach
2. Write helper scripts when you need new capabilities
3. Execute them with bash
4. Use the results

The filesystem is your tool registry.`,
    MaxIterations:    20,
    MaxContextTokens: 128000,
}

result, _ := agent.Run(ctx, "Analyze all Go files and find the most complex function")

A typical execution flow:

Iteration 1: LLM → think("I need to parse Go ASTs. I'll write a Go script.")
Iteration 2: LLM → write_file("analyze.go", "package main...") → bash("go run analyze.go")
Iteration 3: LLM → read_file("analyze.go output") → think("The results show...")
Iteration 4: LLM → "Here's my analysis: ..."  (done)

This is the same pattern used by Claude Code and similar systems: a minimal set of primitives (read, write, edit, bash, think) that's Turing-complete for agent capabilities. No plugin system, no tool registry, no dynamic dispatch — just code and a shell.

See example/autonomous for a runnable example.

State

All state lives in a single flat *State (thread-safe sync.RWMutex + map[string]any). Scoping is done by key prefix convention:

Prefix Purpose Example key
session:: Shared across all agents in a session session::user_id
agent::<name>:: Private to one agent's tools agent::billing::invoice_id
node::<id>:: Output of a graph node node::classifier::output
graph:: Graph-level metadata graph::input
(none) Global config_version

Use the scope constants instead of string literals:

tantra.ScopeSession          // "session::"
tantra.ScopeGraph            // "graph::"
tantra.ScopeAgent("billing") // "agent::billing::"
tantra.ScopeNode("classify") // "node::classify::"
Why a single flat map

The alternative is separate state objects per agent, per node, per session — which creates a coordination problem. When agent A stores an invoice ID and hands off to agent B, how does B get it? Message passing? A shared bus? A merge step?

A single map with prefixed keys is simple to reason about, simple to debug (state.Snapshot() shows everything), and simple to share (WithRunState). The prefix convention gives you isolation when you want it without the machinery of separate stores. ScopedState enforces the convention so tools can't accidentally write to another agent's namespace.

ScopedState

Tools receive a *ScopedState that restricts reads and writes. Writes are always prefixed to the tool's owning agent. Reads use a waterfall: check own scope first, then readable scopes (like session::), then global. This means a billing agent's tool can read session::user_id written by the triage agent without knowing the triage agent exists — it just calls state.Get("user_id").

GetLocal bypasses the waterfall for cases where you need to be certain a value came from your own scope, not inherited from somewhere else.

flowchart TD
    G["state.Get(key)"] --> O{"Own scope\nagent::name::key"}
    O -->|found| R[Return value]
    O -->|miss| S{"Readable scopes\nsession::key"}
    S -->|found| R
    S -->|miss| GL{"Global\nkey"}
    GL -->|found| R
    GL -->|miss| N[Not found]

func myToolFn(ctx context.Context, state *tantra.ScopedState, args MyArgs) (tantra.ToolResult, error) {
    // Write to own scope (auto-prefixed to agent::<name>::)
    state.Set("user_id", "123")

    // Read: checks own scope -> readable scopes (session::) -> global
    v, ok := state.Get("user_id")

    // Read own scope only (no waterfall)
    v, ok = state.GetLocal("user_id")

    // Write to session scope (visible to all agents)
    state.Session().Set("shared_key", "value")

    // Access raw state (for graph keys, etc.)
    input := state.Raw().GetString(tantra.ScopeGraph + "input")

    return tantra.SimpleResult("done"), nil
}
Sharing state between agents
state := tantra.NewState()
state.Set(tantra.ScopeSession+"token", "abc")

// Both agents share the same state
result1, _ := agent1.Run(ctx, "step 1", tantra.WithRunState(state))
result2, _ := agent2.Run(ctx, "step 2", tantra.WithRunState(state))

Multimodal

Send images, audio, and files alongside text using WithParts. The framework uses a provider-agnostic ContentPart type — constructors like ImageURLPart and AudioPart create the right structure, and the OpenAI provider converts them to the API's expected format (data URIs for base64 images, content part arrays for the chat completions endpoint). If you add a custom provider, you control how parts are serialized.

When Parts is set on a message, providers use it instead of Content. This keeps backward compatibility: existing text-only code is unchanged, and Message.TextContent() extracts just the text from multimodal messages if you need it.

// Image from URL
result, _ := agent.Run(ctx, "", tantra.WithParts(
    tantra.TextPart("What's in this image?"),
    tantra.ImageURLPart("https://example.com/photo.jpg", tantra.ImageDetailLow),
))

// Image from local file (reads + base64-encodes automatically)
img, err := tantra.ImageFilePart("/path/to/photo.png")
if err != nil {
    panic(err)
}
result, _ := agent.Run(ctx, "", tantra.WithParts(
    tantra.TextPart("Describe this"),
    img,
))

// Image from base64 data
result, _ := agent.Run(ctx, "", tantra.WithParts(
    tantra.TextPart("OCR this"),
    tantra.ImageBase64Part(b64Data, "image/png", tantra.ImageDetailHigh),
))

// Audio input
result, _ := agent.Run(ctx, "", tantra.WithParts(
    tantra.TextPart("Transcribe this audio"),
    tantra.AudioPart(b64AudioData, "wav"),
))

// File input
result, _ := agent.Run(ctx, "", tantra.WithParts(
    tantra.TextPart("Summarize this document"),
    tantra.FilePart(b64PdfData, "application/pdf", "report.pdf"),
))
Content part constructors
Constructor Content type Fields used
TextPart(text) text Text
ImageURLPart(url, ...detail) image URL, Detail
ImageBase64Part(data, mediaType, ...detail) image Data, MediaType, Detail
ImageFilePart(path, ...detail) image Data, MediaType, Detail
AudioPart(data, format) audio Data, AudioFormat
FilePart(data, mediaType, filename) file Data, MediaType, Filename

ImageDetail options: ImageDetailAuto, ImageDetailLow, ImageDetailHigh. Use ImageDetailLow to reduce token usage when high resolution isn't needed.

Why ContentPart is a flat struct

The alternative is an interface with TextPart, ImagePart, etc. as separate types. But content parts are data, not behavior — they get serialized to JSON for the API. A flat struct with optional fields is easier to construct, marshal, and pass through HTTP endpoints. The constructors (TextPart, ImageURLPart, etc.) provide the ergonomic API; the struct provides the simple data model.

Swarm

Swarm orchestrates multiple agents with dynamic handoffs. The LLM decides when and where to transfer — you define the agents and their capabilities, and the swarm handles routing, state promotion, and conversation threading.

Use Swarm when the routing decision requires understanding the user's intent. A customer support system where "I need a refund" goes to the refund agent but "what's my balance" goes to billing — that's a judgment call the LLM is good at.

triage := &tantra.Agent{
    Name:         "triage",
    Completer:    provider,
    SystemPrompt: "You route customer requests to the right team.",
}

billing := &tantra.Agent{
    Name:         "billing",
    Completer:    provider,
    SystemPrompt: "You handle billing and invoice questions.",
    Tools:        []tantra.Tool{lookupInvoiceTool},
}

refund := &tantra.Agent{
    Name:         "refund",
    Completer:    provider,
    SystemPrompt: "You process refund requests.",
    Tools:        []tantra.Tool{processRefundTool},
}

swarm := &tantra.Swarm{
    Name:        "support",
    Agents:      map[string]*tantra.Agent{"triage": triage, "billing": billing, "refund": refund},
    EntryPoint:  "triage",
    MaxHandoffs: 5, // default: 10
}

result, err := swarm.Run(ctx, "I need a refund for invoice INV-123")
// result.Output       - final response
// result.HandoffChain - e.g. ["triage", "billing", "refund"]
// result.Steps        - per-agent execution details
How it works

The swarm auto-generates tools for each agent:

  • transfer_to_<agent> - hands off control completely. The current agent's scoped state is promoted to session:: scope so the target agent can read it via the normal waterfall lookup. This preserves types — an int stored by triage arrives as an int in billing, not a stringified version.
  • consult_<agent> - asks another agent a question and returns the response. Control stays with the current agent. Useful when you need another agent's expertise without giving up the conversation.
  • list_available_agents - lists all agents and their capabilities. Helps the LLM discover what's available.

Each agent gets a fresh message history on handoff (no context bleed), but the shared *State carries structured data across. The handoff message includes a natural language summary for the LLM, while tools can read the actual structured data from state.

flowchart LR
    U[User Input] --> T[triage]
    T -->|"transfer_to_billing\n(state promoted to session::)"| B[billing]
    B -->|"transfer_to_refund\n(state promoted to session::)"| R[refund]
    R --> O[Result]

    S[(Shared State)]
    T -.->|read/write| S
    B -.->|read/write| S
    R -.->|read/write| S
Restricting handoffs

By default, every agent can transfer to every other agent. Use Handoffs to restrict — useful when you don't want a billing agent accidentally transferring to an admin agent, or when you want to enforce a triage-first flow.

swarm := &tantra.Swarm{
    Agents: map[string]*tantra.Agent{...},
    Handoffs: map[string][]string{
        "triage":  {"billing", "refund"},  // triage can reach billing or refund
        "billing": {"refund"},             // billing can reach refund only
        // refund has no entry -> no outgoing transfers
    },
}

Graph

Graph defines a fixed execution flow with conditional edges. Use Graph when the pipeline structure is known at build time — classify then route, fetch then summarize, validate then process. The LLM still runs inside each node, but the routing between nodes is your code, not the LLM's judgment.

This is the right choice when: the steps are predictable, you need guaranteed execution order, you want to mix LLM nodes with plain functions, or you need fine-grained control over branching conditions.

g := tantra.NewGraph("support-pipeline", tantra.WithMaxIterations(20))

g.AddNode(tantra.NewAgentNode("classify", classifierAgent))
g.AddNode(tantra.NewAgentNode("billing", billingAgent))
g.AddNode(tantra.NewAgentNode("technical", techAgent))
g.AddNode(tantra.NewFunctionNode("log", func(ctx context.Context, state *tantra.State) (string, error) {
    output := state.GetString(tantra.ScopeGraph + "current_output")
    log.Println("Final output:", output)
    return output, nil
}))

g.AddEdge("START", "classify")
g.AddEdge("classify", "billing", tantra.WithConditionFn(func(s *tantra.State) bool {
    return tantra.NodeOutput(s, "classify") == "billing"
}))
g.AddEdge("classify", "technical", tantra.WithConditionFn(func(s *tantra.State) bool {
    return tantra.NodeOutput(s, "classify") == "technical"
}))
g.AddEdge("billing", "log")
g.AddEdge("technical", "log")
g.AddEdge("log", "END")

result, err := g.Run(ctx, "My invoice is wrong")
// result.Output        - final output
// result.ExecutionPath - e.g. ["classify", "billing", "log"]
// result.State         - full state after execution
// result.Success       - whether all nodes succeeded

flowchart TD
    START --> classify
    classify -->|"output == billing"| billing
    classify -->|"output == technical"| technical
    billing --> log
    technical --> log
    log --> END
Swarm vs Graph
Swarm Graph
Routing LLM decides at runtime You decide at build time
When to use Routing requires understanding user intent Steps are predictable, order matters
Nodes Agents only Agents + functions + routers
State flow Auto-promoted on handoff All nodes share state directly
Cycles Bounded by MaxHandoffs Bounded by MaxIterations
Composition Agents can self-call recursively Can embed agents that use Swarm internally

They compose: a graph node can contain an agent that's part of a swarm, or a swarm agent can use a graph internally.

Node types

AgentNode - wraps an *Agent. The graph's state is passed to the agent via WithRunState, so the agent's tools can read/write graph and node state. Each node's output is stored at node::<id>::output automatically.

node := tantra.NewAgentNode("summarizer", agent)

// With custom input transform (default reads graph::current_output)
node := tantra.NewAgentNode("summarizer", agent, tantra.WithInputTransform(func(s *tantra.State) string {
    return "Summarize: " + tantra.NodeOutput(s, "fetcher")
}))

FunctionNode - runs a plain function. No LLM involved. Use for data transforms, validation, logging, API calls, or any deterministic step in the pipeline.

node := tantra.NewFunctionNode("transform", func(ctx context.Context, s *tantra.State) (string, error) {
    input := s.GetString(tantra.ScopeGraph + "current_output")
    return strings.ToUpper(input), nil
})

RouterNode - selects the next node dynamically based on state. Unlike edge conditions (which are evaluated per-edge), a router evaluates all routes and picks the first match. Use for fan-out decisions where you want the routing logic in one place rather than spread across edges.

router := tantra.NewRouterNode("route",
    map[string]func(*tantra.State) bool{
        "billing":   func(s *tantra.State) bool { return tantra.NodeOutput(s, "classify") == "billing" },
        "technical": func(s *tantra.State) bool { return tantra.NodeOutput(s, "classify") == "technical" },
    },
    "fallback", // default route
)
Edge conditions

Edges connect nodes and control flow. By default edges are unconditional (Always). Conditions let you build branching, error handling, and dynamic routing:

g.AddEdge("a", "b")                                           // Always (default)
g.AddEdge("a", "b", tantra.WithCondition(tantra.OnSuccess))   // Only if a succeeded
g.AddEdge("a", "c", tantra.WithCondition(tantra.OnFailure))   // Only if a failed — useful for error recovery
g.AddEdge("a", "d", tantra.WithCondition(tantra.OnToolCall))  // Only if a called tools — detect when LLM took action
g.AddEdge("a", "b", tantra.WithConditionFn(func(s *tantra.State) bool {
    return s.GetString(tantra.ScopeNode("a") + "output") == "ready"
}))
g.AddEdge("a", "", tantra.WithTargetFn(func(s *tantra.State) string {
    return tantra.NodeOutput(s, "a") // dynamic target — node a's output IS the next node ID
}))
g.AddEdge("a", "b", tantra.WithPriority(10)) // higher priority = evaluated first

When multiple edges leave a node, they're evaluated in priority order (descending). The first matching edge wins.

Reading node outputs
// Inside a condition function or input transform:
output := tantra.NodeOutput(state, "classify") // reads node::classify::output

HTTP server

Serve agents over HTTP with zero configuration. The server is intentionally minimal — CORS headers, JSON in/out, health check, SSE streaming. It's meant to get you from "agent works in Go" to "agent works over HTTP" in one line. For production, use Handler and add your own auth, rate limiting, and middleware.

tantra.ListenAndServe(":8080", agent1, agent2, agent3)

Or use Handler with your own server/middleware:

mux := http.NewServeMux()
mux.Handle("/ai/", http.StripPrefix("/ai", tantra.Handler(agent1, agent2)))
http.ListenAndServe(":8080", mux)
Endpoints

GET /health

{"status": "ok", "agents": ["assistant", "researcher"]}

POST /{agent}/run

Text input:

{"message": "What is Go?"}

Multimodal input:

{
  "parts": [
    {"type": "text", "text": "Describe this image"},
    {"type": "image", "url": "https://example.com/photo.jpg", "detail": "low"}
  ]
}

Response:

{
  "output": "Go is a statically typed...",
  "prompt_tokens": 150,
  "output_tokens": 200,
  "tool_calls": 0,
  "duration_ms": 1234,
  "cost": 0.0035
}

POST /{agent}/stream

Same request format as /run. Returns Server-Sent Events (SSE):

data: {"type":"token","content":"Go ","agent":"assistant"}

data: {"type":"token","content":"is a ","agent":"assistant"}

data: {"type":"token","content":"programming language.","agent":"assistant"}

data: {"type":"done","agent":"assistant","data":{"output":"Go is a programming language.","iterations":1,"tool_calls":0,"duration_ms":842}}

See the Streaming section for the full event reference.

Provider

OpenAI (built-in)

The built-in provider uses the official openai-go SDK. WithBaseURL makes it work with any OpenAI-compatible API — Azure OpenAI, LiteLLM proxy, vLLM, Ollama, etc. No need for a multi-provider abstraction layer; the Completer interface already is that layer.

// Default: reads OPENAI_API_KEY env, uses gpt-4o
provider := tantra.NewOpenAI()

// Custom configuration
provider := tantra.NewOpenAI(
    tantra.WithModel("gpt-4o-mini"),
    tantra.WithAPIKey("sk-..."),
    tantra.WithBaseURL("https://my-proxy.example.com/v1"), // for Azure, LiteLLM, etc.
    tantra.WithCost(0.00015, 0.0006), // input/output cost per 1K tokens
)
Custom provider

Completer is deliberately one method. It takes messages and tool schemas, returns a response. No session management, no streaming requirements, no configuration — just the completion call. This makes it trivial to implement for any LLM backend or to wrap for testing.

type Completer interface {
    Complete(ctx context.Context, messages []Message, tools []map[string]any) (*Response, error)
}

Optional interfaces for additional capabilities. Implement these only if your provider supports them — the framework detects them at runtime:

// Streaming support
type Streamer interface {
    Stream(ctx context.Context, messages []Message, tools []map[string]any) <-chan Event
}

// Cost tracking
type CostEstimator interface {
    InputCostPer1K() float64
    OutputCostPer1K() float64
}

Tool schemas use the OpenAI function-calling format because it's the de facto standard — most providers accept it or have a trivial mapping:

[{
  "type": "function",
  "function": {
    "name": "get_weather",
    "description": "Get weather for a city",
    "parameters": {
      "type": "object",
      "properties": {"city": {"type": "string", "description": "City name"}},
      "required": ["city"]
    }
  }
}]

Observability

Every decision point inside the framework — LLM calls, tool executions, handoffs, node transitions — is invisible from the outside. You get a Result with the final output, but no visibility into why the agent called tool A before tool B, how many LLM round-trips a swarm handoff took, or which graph edge was traversed. Logging inside your tools only shows your code, not the framework's decisions.

Observer is a single callback interface that receives structured events at every meaningful point. Set it once, and it flows through context — from swarm to agent, from graph to agent node to agent, from agent to tool chain. No manual plumbing.

type Observer interface {
    Handle(ObserveEvent)
}
Attaching an observer
// On an agent run
result, _ := agent.Run(ctx, "hello", tantra.WithObserver(myObserver))

// On a graph (all nodes inherit it via context)
g := tantra.NewGraph("pipeline", tantra.WithGraphObserver(myObserver))

// On a swarm (all agents inherit it via context)
swarm := &tantra.Swarm{
    Agents:   agents,
    Observer: myObserver,
}
Event types

14 event types covering all three orchestrators:

Event Emitted by When
agent.start / agent.done Agent Wraps the full Agent.Run call
llm.start / llm.done Agent Around each Completer.Complete call
tool.start / tool.done Agent Around each tool execution (including chained tools)
swarm.start / swarm.done Swarm Wraps the swarm orchestration loop
swarm.handoff Swarm When an agent transfers to another
graph.start / graph.done Graph Wraps graph execution
graph.node.start / graph.node.done Graph Around each node execution
graph.edge Graph When an edge is traversed (includes condition name)

Each event carries SpanID and ParentID forming a tree — agent spans contain LLM and tool spans, swarm spans contain agent spans, graph spans contain node spans. This gives you full causality without an external tracing library.

flowchart TD
    SW["swarm.start"] --> A1["agent.start (triage)"]
    A1 --> L1["llm.start / llm.done"]
    A1 --> T1["tool.start / tool.done"]
    SW --> H["swarm.handoff"]
    SW --> A2["agent.start (billing)"]
    A2 --> L2["llm.start / llm.done"]
    A2 --> T2["tool.start / tool.done"]
    SW --> SD["swarm.done"]
Example: simple logger
observer := tantra.ObserverFunc(func(e tantra.ObserveEvent) {
    if e.Error != nil {
        log.Printf("[%s] span=%d parent=%d ERROR: %v", e.Type, e.SpanID, e.ParentID, e.Error)
        return
    }
    log.Printf("[%s] span=%d parent=%d %v", e.Type, e.SpanID, e.ParentID, e.Data)
})

result, _ := agent.Run(ctx, "hello", tantra.WithObserver(observer))
Example: OpenTelemetry bridge

The framework deliberately has zero tracing dependencies. If you want OTel, bridge it:

observer := tantra.ObserverFunc(func(e tantra.ObserveEvent) {
    _, span := tracer.Start(ctx, string(e.Type))
    span.SetAttributes(attribute.String("agent", e.Agent))
    for k, v := range e.Data {
        span.SetAttributes(attribute.String(k, fmt.Sprint(v)))
    }
    if e.Error != nil {
        span.RecordError(e.Error)
    }
    if e.Duration > 0 {
        span.End()
    }
})
Why context-based propagation

The alternative is explicit fields: Agent.Observer, Swarm.Observer, Graph.Observer, and then you need to plumb the observer from swarm to each wrapped agent, from graph to each agent node, etc. Context-based propagation means setting an observer on a swarm automatically makes it available to every agent that runs inside that swarm — no configuration per agent. If you need to override it for a specific agent, WithObserver takes priority over the inherited context observer.

Streaming

Agent.Run blocks until the full response is ready — fine for backend pipelines, but for user-facing applications it means seconds of blank screen. Stream() returns a channel of StreamEvent values as the agent works, so you can render tokens as they arrive, show tool call progress, and track handoffs in real time.

All three orchestrators support streaming: Agent.Stream, Swarm.Stream, and Graph.Stream. They mirror their Run counterparts in setup and behavior — same options, same state handling, same observer events firing alongside. The difference is output delivery: events arrive incrementally instead of all at once.

Agent.Stream
ch := agent.Stream(ctx, "Explain quantum computing", tantra.WithRunState(state))

for evt := range ch {
    switch evt.Type {
    case "token":
        fmt.Print(evt.Content) // render as it arrives
    case "tool_call.start":
        fmt.Printf("\n[calling %s...]\n", evt.Data["tool"])
    case "tool_call.done":
        fmt.Printf("[done: %s]\n", evt.Data["output"])
    case "done":
        fmt.Printf("\n--- completed in %vms ---\n", evt.Data["duration_ms"])
    case "error":
        log.Fatal(evt.Err)
    }
}

If the Completer implements Streamer, tokens arrive as the LLM generates them. If it doesn't, Stream falls back to Complete and emits the full response as a single "token" event. Callers don't need to check provider capabilities.

Swarm.Stream
ch := swarm.Stream(ctx, "I need a refund")

for evt := range ch {
    switch evt.Type {
    case "agent.start":
        fmt.Printf("--- %s ---\n", evt.Agent)
    case "token":
        fmt.Print(evt.Content)
    case "handoff":
        fmt.Printf("\n[handoff: %s -> %s (%s)]\n", evt.Data["from"], evt.Data["to"], evt.Data["reason"])
    case "done":
        break
    }
}

Swarm streaming emits agent.start/agent.done around each agent's turn, forwards all agent events with the Agent field set, and emits handoff events when control transfers.

Graph.Stream
ch := g.Stream(ctx, "Process this request")

for evt := range ch {
    switch evt.Type {
    case "node.start":
        fmt.Printf("[node %s (%s)]\n", evt.Data["node"], evt.Data["type"])
    case "token":
        fmt.Print(evt.Content)
    case "node.done":
        fmt.Printf("\n[node %s done in %vms]\n", evt.Data["node"], evt.Data["duration_ms"])
    case "edge":
        fmt.Printf("[%s -> %s]\n", evt.Data["from"], evt.Data["to"])
    case "done":
        break
    }
}

AgentNodes are streamed — you get live tokens from the LLM inside each node. FunctionNodes and RouterNodes execute synchronously and emit node.start/node.done without token events.

Event reference

Agent events:

Event Fields When
token Content, Agent Each token from the LLM
tool_call.start Data.tool, Data.args Before tool execution
tool_call.done Data.tool, Data.output, Data.duration_ms After tool execution
done Data.output, Data.iterations, Data.tool_calls, Data.duration_ms Agent finished
error Error, Err Error occurred

Swarm adds:

Event Fields When
agent.start Agent Agent begins its turn
agent.done Agent, Data.output Agent finishes its turn
handoff Data.from, Data.to, Data.reason Control transfers

Graph adds:

Event Fields When
node.start Data.node, Data.type Node begins
node.done Data.node, Data.output, Data.success, Data.duration_ms Node finishes
edge Data.from, Data.to, Data.condition Edge traversed
Why StreamEvent is separate from Observer events

StreamEvent and ObserveEvent serve different audiences. StreamEvent is what end users and UIs consume — tokens for rendering, tool progress for status indicators, done events for completion. ObserveEvent is what developers consume — span IDs for tracing, parent-child relationships for causality, timing data for performance analysis. Both fire during streaming; they don't interfere.

The channel API (<-chan StreamEvent) matches the provider-level Streamer pattern and works naturally with for range, select, and Go concurrency primitives. No callbacks, no event bus, no framework-specific abstractions.

MCP (Model Context Protocol)

Connect to external tool servers — databases, file systems, APIs, IDEs — without writing Go wrappers. MCP is the standard protocol for LLM tool interop. MCPClient connects to an MCP server, discovers its tools, and returns []Tool that work with Agent, Swarm, Graph, and Stream unchanged.

// Stdio transport (launches subprocess)
mc, err := tantra.NewMCPStdio("npx", nil, "-y", "@modelcontextprotocol/server-filesystem", "/tmp")
if err != nil {
    panic(err)
}
defer mc.Close()

// HTTP transport
mc, err := tantra.NewMCPHTTP("http://localhost:8080/mcp")
if err != nil {
    panic(err)
}
defer mc.Close()

// MCP tools are just tantra Tools — use them anywhere
agent := &tantra.Agent{
    Name:      "assistant",
    Completer: tantra.NewOpenAI(),
    Tools:     mc.Tools(),
}

result, _ := agent.Run(ctx, "List the files in /tmp")
Mixing MCP and local tools
agent := &tantra.Agent{
    Completer: provider,
    Tools:     append(mc.Tools(), myLocalTool, anotherTool),
}
Multiple MCP servers
files, _ := tantra.NewMCPStdio("npx", nil, "-y", "@modelcontextprotocol/server-filesystem", "/tmp")
defer files.Close()

db, _ := tantra.NewMCPHTTP("http://localhost:3000/mcp")
defer db.Close()

agent := &tantra.Agent{
    Completer: provider,
    Tools:     append(files.Tools(), db.Tools()...),
}
How it works

NewMCPStdio / NewMCPHTTP initialize the MCP connection, call tools/list to discover available tools, and convert each MCP tool definition into a tantra Tool. Schema mapping is automatic — MCP's inputSchema (JSON Schema) is wrapped in the OpenAI function-calling envelope that tantra uses internally.

When a tool is executed, the mcpTool adapter sends a tools/call request to the MCP server and extracts the result. Text content passes through verbatim. Non-text content (images, audio, resources) gets descriptive markers like [image: image/png].

Error handling: MCP protocol errors (connection failure, JSON-RPC errors) become Go errors and break the agent loop. Tool execution errors (isError: true in MCP) are returned as ToolResult.Output text so the LLM can see them and self-correct — matching how tantra handles local tool errors.

MCPClient API
mc.Tools()      // []Tool — discovered tools, ready to use
mc.ServerName() // string — name reported by the MCP server
mc.Close()      // error — shut down connection (important for stdio servers)

Skills

Skills package domain-specific knowledge and tools into reusable directories. The format follows the standard SKILL.md convention used by OpenAI Codex and Anthropic Claude — a YAML frontmatter with name and description, followed by markdown instructions.

my-skill/
  SKILL.md
  references/         # optional — auto-generates tools for on-demand access
    api-spec.md
    examples.txt
SKILL.md format
---
name: code-review
description: Reviews code for bugs, style issues, and security vulnerabilities
---

# Code Review

When reviewing code:
- Check for common security vulnerabilities (SQL injection, XSS)
- Verify error handling covers edge cases
- Look for performance bottlenecks in hot paths
Loading from disk
// Load a single skill
skill, err := tantra.LoadSkill("./skills/code-review")
if err != nil {
    panic(err)
}

// Load all skills from a directory
skills, err := tantra.LoadSkills("./skills")
if err != nil {
    panic(err)
}

// Attach to an agent
agent := &tantra.Agent{
    Name:         "assistant",
    Completer:    provider,
    SystemPrompt: "You are a helpful assistant.",
    Skills:       skills,
}

When attached, skill instructions are appended to the agent's system prompt and skill tools are merged with the agent's tools. Agent tools take precedence on name collision.

Programmatic skills

Skills don't require a directory — construct them in code for testing or dynamic configuration:

skill := &tantra.Skill{
    Name:         "sql-expert",
    Description:  "PostgreSQL query optimization",
    Instructions: "When writing SQL queries, always use parameterized queries...",
    Tools:        []tantra.Tool{explainTool, analyzeTool},
}

agent.Skills = []*tantra.Skill{skill}
Reference files and progressive disclosure

If a skill directory contains a references/ subdirectory, LoadSkill auto-generates two tools:

  • <name>_list_references — lists available reference files
  • <name>_read_reference — reads a specific file by name

This implements progressive disclosure: reference content is loaded on demand by the LLM, not stuffed into the system prompt. An agent with 10 skills and dozens of reference files only loads what it needs per conversation.

skill, _ := tantra.LoadSkill("./skills/api-client")
// skill.Tools contains [api-client_list_references, api-client_read_reference]

// Add your own tools alongside the auto-generated ones
skill.Tools = append(skill.Tools, myCustomTool)
How it works

When an agent runs with skills attached:

  1. effectiveSystemPrompt() appends each skill's instructions to the base system prompt, separated by ## Skill: <name> headers
  2. effectiveTools() merges agent tools with skill tools, deduplicating by name (agent tools win)
  3. Both methods are used by buildInitialMessages and buildToolData, so skills work transparently with Run, Stream, Resume, and through Swarm wrapping

Zero overhead when no skills are attached — the methods return the base system prompt and tools directly.

Checkpointing

Agent runs can fail mid-execution — process crashes, network timeouts, deploys. Checkpointing saves execution state at clean boundaries so runs can resume from where they left off instead of restarting from scratch.

All three orchestrators support checkpointing: Agent, Swarm, and Graph. Provide a CheckpointStore and the framework saves state after each iteration boundary automatically. Resume with the run ID.

Agent
store := tantra.NewMemoryCheckpointStore() // or your own Postgres/Redis store

result, err := agent.Run(ctx, "Research RLHF papers",
    tantra.WithCheckpointStore(store),
    tantra.WithRunID("run-123"), // optional, auto-generated if omitted
)

// Later, after a crash:
result, err = agent.Resume(ctx, "run-123", store)

Checkpoints are saved after each LLM call + tool execution round. On resume, the agent continues from the last checkpoint with accumulated messages, stats, and state intact.

Swarm
swarm := &tantra.Swarm{
    Agents:          agents,
    EntryPoint:      "triage",
    CheckpointStore: store,
}

result, err := swarm.Run(ctx, "I need a refund")

// Resume after crash — picks up with the correct agent
result, err = swarm.Resume(ctx, runID)

Checkpoints are saved after each agent step and handoff resolution. On resume, the swarm continues with the correct agent, input, handoff chain, and shared state.

Graph
g := tantra.NewGraph("pipeline", tantra.WithGraphCheckpointStore(store))
// ... add nodes and edges ...

result, err := g.Run(ctx, "Process this")

// Resume from the last completed node
result, err = g.Resume(ctx, runID)

Checkpoints are saved after each node executes. On resume, the graph continues from the next node with all previous node outputs preserved in state.

CheckpointStore interface
type CheckpointStore interface {
    Save(ctx context.Context, cp Checkpoint) error
    Load(ctx context.Context, runID string) (Checkpoint, bool, error)
    List(ctx context.Context, runID string) ([]string, error)
    Delete(ctx context.Context, runID string) error
}

MemoryCheckpointStore is included for testing. For production, implement the interface with your preferred storage (Postgres, Redis, S3, etc.). Load returns the latest checkpoint for a run. Delete cleans up after a completed run.

Design notes
  • Zero overhead when not configured. No checkpoint store = no checkpointing, no behavior change.
  • Best-effort saves. Checkpoint save failures don't fail the run — checkpointing is a safety net, not a critical path.
  • DurationMS on resume measures the resumed segment only. Tracking cumulative wall-clock time across process restarts is misleading.
  • JSON-serializable. Checkpoint uses map[string]any for state data and JSON tags throughout. State values round-tripped through JSON follow Go's standard behavior (integers become float64).
flowchart LR
    R["Run(input)"] --> I1["Iteration 1\nLLM + Tools"]
    I1 -->|save checkpoint| I2["Iteration 2\nLLM + Tools"]
    I2 -->|save checkpoint| I3["Iteration 3\nLLM + Tools"]
    I3 -->|crash| X[ ]
    X -.->|"Resume(runID)"| I3R["Iteration 3\n(replayed from checkpoint)"]
    I3R --> I4["Iteration 4\nLLM → Result"]
    style X fill:none,stroke:none

Context compaction

As an agent iterates or a multi-turn conversation grows, messages accumulate without bound — the entire history is sent on every LLM call. For long-running agents this eventually exceeds the model's context window. Even before failure, sending a massive history wastes tokens and money.

MaxContextTokens enables automatic compaction: when the estimated token count exceeds the limit, older messages are summarized by the LLM and replaced with a concise summary. Recent messages are preserved verbatim.

agent := &tantra.Agent{
    Completer:        tantra.NewOpenAI(tantra.WithModel("gpt-4o")),
    MaxContextTokens: 128000, // compact when approaching this limit
}

When compaction triggers:

  1. The system prompt is always preserved
  2. Recent messages (up to 50% of the token budget) are kept verbatim
  3. Older messages are summarized into a single [Previous conversation summary] message
  4. Tool call/result pairs are never split — boundaries are kept clean
flowchart LR
    S[System] --> OLD["Older messages\n(summarized)"]
    OLD --> SUM["[Summary]"]
    SUM --> REC["Recent messages\n(preserved)"]
    REC --> LLM[LLM Call]
  • Zero overhead when not set. MaxContextTokens: 0 (default) means no compaction, no token estimation, no behavior change.
  • Best-effort. If the summary LLM call fails, the original messages are sent unchanged.
  • Token accounting. The summary call's tokens are counted in Result.PromptTokens / Result.OutputTokens (it's a real cost), but not in Iterations or ToolCalls.
  • Works everywhere. Transparent with Run, Stream, Resume, and WithMessages (multi-turn).

Observer events compact.start / compact.done are emitted when compaction occurs, with message counts and token estimates.

File map

tantra-go/
  types.go       Message, ContentPart, Role, ToolCall, Response, Result, Event, StreamEvent
  content.go     TextPart, ImageURLPart, ImageBase64Part, AudioPart, FilePart
  state.go       State, ScopedState, ToolResult, scope constants
  tool.go        Tool interface, NewTool[T], SimpleTool, Param
  builtins.go    ReadTool, WriteTool, EditTool, BashTool, CodeTools
  skill.go       Skill, LoadSkill, LoadSkills, parseSKILLMD, reference tools, effectiveSystemPrompt, effectiveTools
  compact.go     Context compaction: estimateTokens, compactMessages, formatMessagesForSummary
  checkpoint.go  Checkpoint, CheckpointStore, MemoryCheckpointStore, Agent.Resume, Swarm.Resume, Graph.Resume
  agent.go       Agent, RunOption, Agent.Run, Agent.Stream, consumeProviderStream
  observe.go     Observer, ObserveEvent, EventType constants, span tracking, context propagation
  provider.go    Completer, Streamer, CostEstimator interfaces
  openai.go      OpenAI provider implementation
  swarm.go       Swarm, Swarm.Run, Swarm.Stream, transfer/consult tools, handoff detection
  graph.go       Graph, Graph.Run, Graph.Stream, AgentNode, FunctionNode, RouterNode, Edge
  mcp.go         MCPClient, NewMCPStdio, NewMCPHTTP, mcpTool adapter
  serve.go       HTTP server: POST /{name}/run, POST /{name}/stream (SSE)
  errors.go      Sentinel errors (ErrNoCompleter, ErrMaxIterations, etc.)
  example/       Standalone runnable examples (main.go per subdirectory)

Documentation

Overview

Package tantra provides a minimal, type-safe agent framework for LLMs.

The core abstraction is Agent, which wraps an LLM provider (Completer) and optional Tool functions. For multi-agent workflows, use Swarm for dynamic handoffs or Graph for predefined execution flows.

Example: 

agent := &tantra.Agent{
    Name:      "assistant",
    Completer: tantra.NewOpenAI(tantra.WithAPIKey(key)),
}
result, err := agent.Run(ctx, "Hello!")
Example (Basic)

This example shows how to create a simple agent. 

package main

import (
	"context"
	"fmt"

	"github.com/tantra-run/tantra-go"
)

func main() {
	agent := &tantra.Agent{
		Name:         "assistant",
		Completer:    &mockCompleter{response: "Hello, human!"},
		SystemPrompt: "You are a helpful assistant.",
	}

	result, err := agent.Run(context.Background(), "Hi there")
	if err != nil {
		panic(err)
	}

	fmt.Println(result.Output)
}

// mockCompleter is a simple completer for examples.
type mockCompleter struct {
	response string
}

func (m *mockCompleter) Complete(ctx context.Context, msgs []tantra.Message, tools []map[string]any) (*tantra.Response, error) {
	return &tantra.Response{Content: m.response}, nil
}

Output:
Hello, human!
Example (Tool)

This example shows how to create a type-safe tool. 

package main

import (
	"context"
	"fmt"

	"github.com/tantra-run/tantra-go"
)

func main() {
	type WeatherArgs struct {
		City string `json:"city" desc:"City name" required:"true"`
	}

	tool := tantra.NewTool("get_weather", "Get weather for a city",
		func(ctx context.Context, state *tantra.ScopedState, args WeatherArgs) (tantra.ToolResult, error) {
			return tantra.SimpleResult(fmt.Sprintf("Weather in %s: Sunny, 72°F", args.City)), nil
		})

	s := tantra.NewState()
	scope := tantra.NewScopedState(s, "test::")
	result, _ := tool.Execute(context.Background(), scope, map[string]any{"city": "Tokyo"})
	fmt.Println(result.Output)
}

Output:
Weather in Tokyo: Sunny, 72°F

Index

Examples

Constants

View Source 
const (
	CheckpointAgent = "agent"
	CheckpointSwarm = "swarm"
	CheckpointGraph = "graph"
)

Checkpoint type constants.

View Source 
const (
	StatusRunning   = "running"
	StatusCompleted = "completed"
	StatusFailed    = "failed"
)

Checkpoint status constants.

View Source 
const (
	ScopeSession = "session::"
	ScopeGraph   = "graph::"
)

Scope prefix constants. Use these instead of string literals.

View Source 
const DefaultMaxGraphIterations = 50

DefaultMaxGraphIterations prevents infinite loops in cyclic graphs.

View Source 
const DefaultMaxHandoffs = 10

DefaultMaxHandoffs is the default limit for agent handoffs.

View Source 
const DefaultMaxIterations = 10

DefaultMaxIterations limits agent loops to prevent runaway costs.

Variables

View Source 
var (
	ErrNoCompleter    = errors.New("no completer configured")
	ErrMaxIterations  = errors.New("max iterations exceeded")
	ErrToolNotFound   = errors.New("tool not found")
	ErrAgentNotFound  = errors.New("agent not found")
	ErrInvalidSession = errors.New("invalid session ID")
)

Sentinel errors returned by tantra functions. Use errors.Is to check for these errors.

Functions

func EstimateCost 

func EstimateCost(c Completer, promptTokens, outputTokens int) float64

EstimateCost calculates the cost of a completion if the provider implements CostEstimator. Returns 0 otherwise.

func Handler 

func Handler(agents ...*Agent) http.Handler

Handler returns an http.Handler for the given agents. Use this with your own server or middleware.

func ListenAndServe 

func ListenAndServe(addr string, agents ...*Agent) error

ListenAndServe starts an HTTP server on the given address.

func NodeOutput 

func NodeOutput(s *State, nodeID string) string

NodeOutput reads a node's output from state.

func ScopeAgent 

func ScopeAgent(name string) string

ScopeAgent returns the prefix for a named agent, e.g. "agent::billing::".

func ScopeNode 

func ScopeNode(name string) string

ScopeNode returns the prefix for a named node, e.g. "node::classifier::".

Types

type Agent 

type Agent struct {
	Name             string    // identifies the agent (required for [Server])
	Completer        Completer // LLM provider (required)
	Tools            []Tool    // available tools (optional)
	SystemPrompt     string    // behavior instructions (optional)
	MaxIterations    int       // loop limit; 0 means DefaultMaxIterations
	MaxContextTokens int       // compact older messages when estimated tokens exceed this; 0 means no compaction
	Skills           []*Skill  // domain-specific knowledge and tools (optional)
}

Agent is an AI agent that uses an LLM to process input and optionally call tools. The zero value is not usable; at minimum, set Agent.Completer.

func (*Agent) Resume 

func (a *Agent) Resume(ctx context.Context, runID string, store CheckpointStore, opts ...RunOption) (*Result, error)

Resume loads the latest checkpoint for the given run and continues execution.

func (*Agent) Run 

func (a *Agent) Run(ctx context.Context, input string, opts ...RunOption) (*Result, error)

Run executes the agent with the given input. It returns when the agent produces a final response or an error occurs. The context can be used for cancellation and timeouts.

func (*Agent) Stream 

func (a *Agent) Stream(ctx context.Context, input string, opts ...RunOption) <-chan StreamEvent

Stream executes the agent with the given input and streams events as they occur. It returns a channel that emits StreamEvent values. The channel is closed when the agent finishes or an error occurs. If the Completer does not implement Streamer, it falls back to [Complete] and emits the full response as a single token.

type AgentCheckpoint 

type AgentCheckpoint struct {
	AgentName    string    `json:"agent_name"`
	Messages     []Message `json:"messages"`
	Iteration    int       `json:"iteration"`
	PromptTokens int       `json:"prompt_tokens"`
	OutputTokens int       `json:"output_tokens"`
	ToolCalls    int       `json:"tool_calls"`
	ToolResults  []string  `json:"tool_results"`
}

AgentCheckpoint captures agent loop state at a clean boundary.

type AgentNode 

type AgentNode struct {
	// contains filtered or unexported fields
}

AgentNode wraps an Agent for graph execution. The graph's State is passed to the agent via WithRunState, so the agent's tools can read/write graph and node state.

func NewAgentNode 

func NewAgentNode(id string, agent *Agent, opts ...AgentNodeOption) *AgentNode

NewAgentNode creates a node that runs an agent.

func (*AgentNode) Execute 

func (n *AgentNode) Execute(ctx context.Context, state *State) (string, bool, bool, error)

Execute runs the agent with the current graph state.

func (*AgentNode) ID 

func (n *AgentNode) ID() string

ID returns the node identifier.

type AgentNodeOption 

type AgentNodeOption func(*AgentNode)

AgentNodeOption configures an AgentNode.

func WithInputTransform 

func WithInputTransform(fn func(*State) string) AgentNodeOption

WithInputTransform sets a function to derive input from state.

type Checkpoint 

type Checkpoint struct {
	ID        string         `json:"id"`
	RunID     string         `json:"run_id"`
	Seq       int            `json:"seq"`
	Type      string         `json:"type"`
	Status    string         `json:"status"`
	CreatedAt time.Time      `json:"created_at"`
	StateData map[string]any `json:"state_data"`

	Agent *AgentCheckpoint `json:"agent,omitempty"`
	Swarm *SwarmCheckpoint `json:"swarm,omitempty"`
	Graph *GraphCheckpoint `json:"graph,omitempty"`
}

Checkpoint captures execution state at a clean boundary. It is JSON-serializable and self-contained for resumption.

type CheckpointStore 

type CheckpointStore interface {
	// Save persists a checkpoint. If a checkpoint with the same ID exists, it is overwritten.
	Save(ctx context.Context, cp Checkpoint) error

	// Load retrieves the latest checkpoint for a run.
	// Returns the checkpoint and true, or a zero Checkpoint and false if not found.
	Load(ctx context.Context, runID string) (Checkpoint, bool, error)

	// List returns all checkpoint IDs for a run, ordered by sequence number.
	List(ctx context.Context, runID string) ([]string, error)

	// Delete removes all checkpoints for a run.
	Delete(ctx context.Context, runID string) error
}

CheckpointStore persists checkpoints for durable execution. Implementations must be safe for concurrent use.

type Completer 

type Completer interface {
	// Complete sends messages to the LLM and returns a response.
	// The tools parameter contains JSON schemas for available tools.
	Complete(ctx context.Context, messages []Message, tools []map[string]any) (*Response, error)
}

Completer is the core interface for LLM providers. Implement this interface to add support for new LLM backends.

type ContentPart 

type ContentPart struct {
	Type        ContentType `json:"type"`
	Text        string      `json:"text,omitempty"`         // ContentText
	URL         string      `json:"url,omitempty"`          // ContentImage (URL source)
	Data        string      `json:"data,omitempty"`         // base64-encoded binary
	MediaType   string      `json:"media_type,omitempty"`   // MIME type for Data
	Detail      ImageDetail `json:"detail,omitempty"`       // ContentImage resolution
	AudioFormat string      `json:"audio_format,omitempty"` // "wav", "mp3"
	Filename    string      `json:"filename,omitempty"`     // ContentFile
}

ContentPart is one piece of a multimodal message. Use the constructors TextPart, ImageURLPart, ImageBase64Part, AudioPart, and FilePart to create parts.

func AudioPart 

func AudioPart(data, format string) ContentPart

AudioPart creates an audio content part from base64-encoded data. Format should be "wav" or "mp3".

func FilePart 

func FilePart(data, mediaType, filename string) ContentPart

FilePart creates a file content part from base64-encoded data.

func ImageBase64Part 

func ImageBase64Part(data, mediaType string, detail ...ImageDetail) ContentPart

ImageBase64Part creates an image content part from base64-encoded data.

func ImageFilePart 

func ImageFilePart(path string, detail ...ImageDetail) (ContentPart, error)

ImageFilePart reads an image file and creates a base64-encoded content part. Supported formats: JPEG, PNG, GIF, WebP.

func ImageURLPart 

func ImageURLPart(url string, detail ...ImageDetail) ContentPart

ImageURLPart creates an image content part from a URL. The optional detail parameter controls resolution ("auto", "low", "high").

func TextPart 

func TextPart(text string) ContentPart

TextPart creates a text content part.

type ContentType 

type ContentType string

ContentType identifies the kind of content in a ContentPart. 

const (
	ContentText  ContentType = "text"
	ContentImage ContentType = "image"
	ContentAudio ContentType = "audio"
	ContentFile  ContentType = "file"
)

type CostEstimator 

type CostEstimator interface {
	InputCostPer1K() float64
	OutputCostPer1K() float64
}

CostEstimator is an optional interface for cost calculation. Implement this to enable automatic cost tracking in Result.

type Edge 

type Edge struct {
	Source      string
	Target      string
	Condition   EdgeCondition
	ConditionFn func(*State) bool   // For Custom condition
	TargetFn    func(*State) string // For Conditional edges
	Priority    int                 // Higher = evaluated first
}

Edge connects two nodes with optional conditions.

func (*Edge) ResolveTarget 

func (e *Edge) ResolveTarget(state *State) string

ResolveTarget returns the target node ID.

func (*Edge) ShouldTraverse 

func (e *Edge) ShouldTraverse(state *State, success, toolCalled bool) bool

ShouldTraverse checks if this edge should be taken.

type EdgeCondition 

type EdgeCondition int

EdgeCondition defines when an edge should be traversed. 

const (
	// Always traverses the edge unconditionally.
	Always EdgeCondition = iota
	// OnSuccess traverses only if the previous node succeeded.
	OnSuccess
	// OnFailure traverses only if the previous node failed.
	OnFailure
	// OnToolCall traverses only if the previous node called a tool.
	OnToolCall
	// Custom uses a custom condition function.
	Custom
	// Conditional dynamically resolves the target node.
	Conditional
)

type EdgeOption 

type EdgeOption func(*Edge)

EdgeOption configures an Edge.

func WithCondition 

func WithCondition(cond EdgeCondition) EdgeOption

WithCondition sets the edge condition.

func WithConditionFn 

func WithConditionFn(fn func(*State) bool) EdgeOption

WithConditionFn sets a custom condition function.

func WithPriority 

func WithPriority(p int) EdgeOption

WithPriority sets the edge evaluation priority.

func WithTargetFn 

func WithTargetFn(fn func(*State) string) EdgeOption

WithTargetFn sets a dynamic target function (makes edge Conditional).

type Event 

type Event struct {
	Type    string // "token", "tool_call", "tool_result", "complete", "error"
	Content string
	Err     error
}

Event represents a provider-level streaming event (internal).

type EventType 

type EventType string

EventType identifies what happened at an observation point. 

const (
	EventAgentStart     EventType = "agent.start"      // emitted before Agent.Run begins
	EventAgentDone      EventType = "agent.done"       // emitted after Agent.Run completes
	EventLLMStart       EventType = "llm.start"        // emitted before each LLM call
	EventLLMDone        EventType = "llm.done"         // emitted after each LLM call
	EventToolStart      EventType = "tool.start"       // emitted before tool execution
	EventToolDone       EventType = "tool.done"        // emitted after tool execution
	EventSwarmStart     EventType = "swarm.start"      // emitted before swarm orchestration begins
	EventSwarmDone      EventType = "swarm.done"       // emitted after swarm orchestration completes
	EventSwarmHandoff   EventType = "swarm.handoff"    // emitted when an agent transfers to another
	EventGraphStart     EventType = "graph.start"      // emitted before graph execution begins
	EventGraphDone      EventType = "graph.done"       // emitted after graph execution completes
	EventGraphNodeStart EventType = "graph.node.start" // emitted before a graph node executes
	EventGraphNodeDone  EventType = "graph.node.done"  // emitted after a graph node executes
	EventGraphEdge      EventType = "graph.edge"       // emitted when a graph edge is traversed
	EventCompactStart   EventType = "compact.start"    // emitted before context compaction
	EventCompactDone    EventType = "compact.done"     // emitted after context compaction
)

type FunctionNode 

type FunctionNode struct {
	// contains filtered or unexported fields
}

FunctionNode executes a custom function.

func NewFunctionNode 

func NewFunctionNode(id string, fn func(context.Context, *State) (string, error)) *FunctionNode

NewFunctionNode creates a node that runs a function.

func (*FunctionNode) Execute 

func (n *FunctionNode) Execute(ctx context.Context, state *State) (string, bool, bool, error)

Execute runs the node's function.

func (*FunctionNode) ID 

func (n *FunctionNode) ID() string

ID returns the node identifier.

type Graph 

type Graph struct {
	// contains filtered or unexported fields
}

Graph orchestrates workflow execution with conditional edges. All nodes share a single State, and [AgentNode]s pass that state to their agent's tools via WithRunState.

func NewGraph 

func NewGraph(name string, opts ...GraphOption) *Graph

NewGraph creates a graph with the given options.

func (*Graph) AddEdge 

func (g *Graph) AddEdge(source, target string, opts ...EdgeOption) *Graph

AddEdge adds an edge between nodes.

func (*Graph) AddNode 

func (g *Graph) AddNode(node Node) *Graph

AddNode adds a node to the graph.

func (*Graph) Resume 

func (g *Graph) Resume(ctx context.Context, runID string, opts ...RunOption) (*GraphResult, error)

Resume loads the latest checkpoint for the given run and continues execution.

func (*Graph) Run 

func (g *Graph) Run(ctx context.Context, input string, opts ...RunOption) (*GraphResult, error)

Run executes the graph with the given input. A single State is created (or provided via WithRunState) and passed to every node. [AgentNode]s forward this state to their agent's tools.

func (*Graph) SetEntryPoint 

func (g *Graph) SetEntryPoint(nodeID string) *Graph

SetEntryPoint sets the starting node.

func (*Graph) SetFinishPoint 

func (g *Graph) SetFinishPoint(nodeID string) *Graph

SetFinishPoint marks a node as an exit point.

func (*Graph) Stream 

func (g *Graph) Stream(ctx context.Context, input string, opts ...RunOption) <-chan StreamEvent

Stream executes the graph and streams events as they occur. It returns a channel that emits StreamEvent values. The channel is closed when the graph finishes. AgentNodes are streamed; other nodes execute synchronously.

func (*Graph) Validate 

func (g *Graph) Validate() []string

Validate checks the graph structure and returns errors.

type GraphCheckpoint 

type GraphCheckpoint struct {
	GraphName     string   `json:"graph_name"`
	CurrentNode   string   `json:"current_node"`
	Iteration     int      `json:"iteration"`
	ExecutionPath []string `json:"execution_path"`
	NodesExecuted []string `json:"nodes_executed"`
	Success       bool     `json:"success"`
}

GraphCheckpoint captures graph loop state at a clean boundary.

type GraphOption 

type GraphOption func(*Graph)

GraphOption configures a Graph.

func WithGraphCheckpointStore 

func WithGraphCheckpointStore(store CheckpointStore) GraphOption

WithGraphCheckpointStore enables checkpointing for graph execution.

func WithGraphObserver 

func WithGraphObserver(o Observer) GraphOption

WithGraphObserver attaches an observer that receives events during graph execution.

func WithMaxIterations 

func WithMaxIterations(max int) GraphOption

WithMaxIterations sets the maximum iterations for cyclic graphs.

type GraphResult 

type GraphResult struct {
	Output        string
	State         *State
	ExecutionPath []string
	NodesExecuted []string
	Iterations    int
	Success       bool
	DurationMS    int64
}

GraphResult contains the outcome of a graph run.

type HealthResponse 

type HealthResponse struct {
	Status string   `json:"status"`
	Agents []string `json:"agents"`
}

HealthResponse is returned by GET /health.

type ImageDetail 

type ImageDetail string

ImageDetail controls image resolution for vision models. 

const (
	ImageDetailAuto ImageDetail = "auto"
	ImageDetailLow  ImageDetail = "low"
	ImageDetailHigh ImageDetail = "high"
)

type MCPClient 

type MCPClient struct {
	// contains filtered or unexported fields
}

MCPClient connects to an MCP server, discovers its tools, and exposes them as tantra Tool implementations. Tools returned by MCPClient.Tools work seamlessly with Agent, Swarm, Graph, and Agent.Stream.

The client must be closed when no longer needed to release resources (especially for stdio servers, which run as child processes).

func NewMCPHTTP 

func NewMCPHTTP(baseURL string) (*MCPClient, error)

NewMCPHTTP connects to an MCP server via streamable HTTP transport. The baseURL should be the root URL of the MCP server endpoint.

The client initializes the connection and discovers all available tools. Call MCPClient.Close when done.

func NewMCPStdio 

func NewMCPStdio(command string, env []string, args ...string) (*MCPClient, error)

NewMCPStdio connects to an MCP server via stdio transport. The command is started as a child process with the given arguments. Environment variables are specified as "KEY=VALUE" strings; pass nil to inherit the current process environment.

The client initializes the connection and discovers all available tools. Call MCPClient.Close when done.

func (*MCPClient) Close 

func (mc *MCPClient) Close() error

Close shuts down the MCP connection and releases resources.

func (*MCPClient) ServerName 

func (mc *MCPClient) ServerName() string

ServerName returns the name reported by the MCP server during initialization.

func (*MCPClient) Tools 

func (mc *MCPClient) Tools() []Tool

Tools returns the discovered MCP tools as tantra Tool implementations. Do not use after calling MCPClient.Close.

type MemoryCheckpointStore 

type MemoryCheckpointStore struct {
	// contains filtered or unexported fields
}

MemoryCheckpointStore is an in-memory CheckpointStore for testing. Checkpoints are lost on process restart.

func NewMemoryCheckpointStore 

func NewMemoryCheckpointStore() *MemoryCheckpointStore

NewMemoryCheckpointStore creates an in-memory checkpoint store.

func (*MemoryCheckpointStore) Delete 

func (m *MemoryCheckpointStore) Delete(_ context.Context, runID string) error

Delete removes all checkpoints for a run.

func (*MemoryCheckpointStore) List 

func (m *MemoryCheckpointStore) List(_ context.Context, runID string) ([]string, error)

List returns all checkpoint IDs for a run.

func (*MemoryCheckpointStore) Load 

func (m *MemoryCheckpointStore) Load(_ context.Context, runID string) (Checkpoint, bool, error)

Load returns the latest checkpoint for a run. Returns false if none exist.

func (*MemoryCheckpointStore) Save 

func (m *MemoryCheckpointStore) Save(_ context.Context, cp Checkpoint) error

Save persists a checkpoint, replacing any existing one with the same ID.

type Message 

type Message struct {
	Role       Role          `json:"role"`
	Content    string        `json:"content,omitempty"`
	Parts      []ContentPart `json:"parts,omitempty"`
	ToolCallID string        `json:"tool_call_id,omitempty"`
	ToolCalls  []ToolCall    `json:"tool_calls,omitempty"`
}

Message represents a message in the conversation history. If [Parts] is non-empty, providers use it instead of [Content].

func (Message) IsMultimodal 

func (m Message) IsMultimodal() bool

IsMultimodal returns true if the message has non-text content parts.

func (Message) TextContent 

func (m Message) TextContent() string

TextContent returns the text content of the message. If Parts is set, it concatenates all text parts. Otherwise returns Content.

type Node 

type Node interface {
	// ID returns the unique identifier for this node.
	ID() string
	// Execute runs the node and returns (output, success, toolCalled, error).
	Execute(ctx context.Context, state *State) (string, bool, bool, error)
}

Node is the interface for graph nodes. Nodes receive the shared State and can read/write to it freely.

type ObserveEvent 

type ObserveEvent struct {
	Type      EventType
	Agent     string         // agent name (empty for graph-only events)
	SpanID    int64          // unique ID for this span
	ParentID  int64          // parent span (0 = root)
	Data      map[string]any // event-specific payload
	Error     error          // non-nil if the operation failed
	Timestamp time.Time
	Duration  time.Duration // set on "done" events
}

ObserveEvent is emitted at each decision point during execution. Start events are emitted before the operation; done events after. SpanID/ParentID form a tree: agent spans contain llm and tool spans, swarm spans contain agent spans, etc.

type Observer 

type Observer interface {
	Handle(ObserveEvent)
}

Observer receives events during execution. Implementations must be safe for concurrent use.

type ObserverFunc 

type ObserverFunc func(ObserveEvent)

ObserverFunc adapts a plain function to the Observer interface.

func (ObserverFunc) Handle 

func (f ObserverFunc) Handle(e ObserveEvent)

Handle calls the underlying function.

type OpenAI 

type OpenAI struct {
	// contains filtered or unexported fields
}

OpenAI implements Completer, Streamer, and CostEstimator using the OpenAI API.

func NewOpenAI 

func NewOpenAI(opts ...OpenAIOption) *OpenAI

NewOpenAI creates an OpenAI provider. By default, it reads OPENAI_API_KEY from environment and uses gpt-4o.

func (*OpenAI) Complete 

func (o *OpenAI) Complete(ctx context.Context, messages []Message, tools []map[string]any) (*Response, error)

Complete implements the Completer interface.

func (*OpenAI) InputCostPer1K 

func (o *OpenAI) InputCostPer1K() float64

InputCostPer1K implements the CostEstimator interface.

func (*OpenAI) OutputCostPer1K 

func (o *OpenAI) OutputCostPer1K() float64

OutputCostPer1K implements the CostEstimator interface.

func (*OpenAI) Stream 

func (o *OpenAI) Stream(ctx context.Context, messages []Message, tools []map[string]any) <-chan Event

Stream implements the Streamer interface using SSE.

type OpenAIOption 

type OpenAIOption func(*openaiConfig)

OpenAIOption configures an OpenAI provider.

func WithAPIKey 

func WithAPIKey(key string) OpenAIOption

WithAPIKey sets the API key (alternative to OPENAI_API_KEY env var).

func WithBaseURL 

func WithBaseURL(url string) OpenAIOption

WithBaseURL sets a custom base URL (for Azure or compatible APIs).

func WithCost 

func WithCost(inputPer1K, outputPer1K float64) OpenAIOption

WithCost sets the cost per 1K tokens for estimation.

func WithModel 

func WithModel(model string) OpenAIOption

WithModel sets the model to use (default: gpt-4o).

type Param 

type Param struct {
	Name        string
	Type        string // "string", "integer", "number", "boolean"
	Description string
	Required    bool
}

Param describes a tool parameter for SimpleTool.

type Response 

type Response struct {
	Content      string     `json:"content,omitempty"`
	ToolCalls    []ToolCall `json:"tool_calls,omitempty"`
	PromptTokens int        `json:"prompt_tokens"`
	OutputTokens int        `json:"output_tokens"`
}

Response represents a response from an LLM provider.

type Result 

type Result struct {
	Output       string
	Messages     []Message // full conversation history (pass back via WithMessages for multi-turn)
	PromptTokens int
	OutputTokens int
	ToolCalls    int
	Iterations   int
	DurationMS   int64
	Cost         float64
	ToolResults  []string // Results from tool executions (for swarm handoff detection)
}

Result contains the outcome of an agent run.

type Role 

type Role string

Role is the role of a message sender in a conversation. 

const (
	RoleSystem    Role = "system"
	RoleUser      Role = "user"
	RoleAssistant Role = "assistant"
	RoleTool      Role = "tool"
)

type RouterNode 

type RouterNode struct {
	// contains filtered or unexported fields
}

RouterNode routes to different nodes based on conditions.

func NewRouterNode 

func NewRouterNode(id string, routes map[string]func(*State) bool, defaultRoute string) *RouterNode

NewRouterNode creates a router that selects the next node.

func (*RouterNode) Execute 

func (n *RouterNode) Execute(ctx context.Context, state *State) (string, bool, bool, error)

Execute evaluates route conditions and returns the selected target node ID.

func (*RouterNode) ID 

func (n *RouterNode) ID() string

ID returns the node identifier.

type RunOption 

type RunOption func(*runConfig)

RunOption configures a single Run call.

func WithCheckpointStore 

func WithCheckpointStore(store CheckpointStore) RunOption

WithCheckpointStore enables checkpointing for this run. After each iteration boundary, the agent saves its state to the store. Use Agent.Resume to continue a checkpointed run.

func WithMessages 

func WithMessages(msgs []Message) RunOption

WithMessages provides conversation history for multi-turn interactions. The agent prepends its system prompt (if set and not already present), appends the new user message (from input or WithParts), and continues the loop. The final message history is returned in Result.Messages.

func WithObserver 

func WithObserver(o Observer) RunOption

WithObserver attaches an observer that receives events during the run. If the context already carries an observer (e.g. from a Swarm or Graph), this option overrides it for this agent run.

func WithParts 

func WithParts(parts ...ContentPart) RunOption

WithParts sets multimodal content parts for the initial user message. When set, parts are used instead of the plain text input.

func WithRunID 

func WithRunID(id string) RunOption

WithRunID sets the run ID for checkpoint grouping. If not set, a random ID is generated.

func WithRunState 

func WithRunState(state *State) RunOption

WithRunState provides an existing State to share across agents or runs.

type RunRequest 

type RunRequest struct {
	Message string        `json:"message"`
	Parts   []ContentPart `json:"parts,omitempty"`
}

RunRequest is the request body for POST /{name}/run. Provide either Message for text-only or Parts for multimodal input.

type RunResponse 

type RunResponse struct {
	Output       string  `json:"output,omitempty"`
	PromptTokens int     `json:"prompt_tokens,omitempty"`
	OutputTokens int     `json:"output_tokens,omitempty"`
	ToolCalls    int     `json:"tool_calls,omitempty"`
	DurationMS   int64   `json:"duration_ms,omitempty"`
	Cost         float64 `json:"cost,omitempty"`
	Error        string  `json:"error,omitempty"`
}

RunResponse is returned by POST /{name}/run.

type ScopedState 

type ScopedState struct {
	// contains filtered or unexported fields
}

ScopedState provides a restricted view of a State. Writes go to a specific prefix; reads check own prefix, then readable prefixes, then global (no prefix).

func NewScopedState 

func NewScopedState(state *State, prefix string, readablePrefixes ...string) *ScopedState

NewScopedState creates a scoped view of a state.

func (*ScopedState) Get 

func (ss *ScopedState) Get(key string) (any, bool)

Get reads a value, checking own scope -> readable scopes -> global.

func (*ScopedState) GetLocal 

func (ss *ScopedState) GetLocal(key string) (any, bool)

GetLocal reads a value from this scope only, without waterfall lookup. Use this when you need to be certain the value came from your own scope.

func (*ScopedState) GetString 

func (ss *ScopedState) GetString(key string) string

GetString retrieves a string value from scoped lookup.

func (*ScopedState) Prefix 

func (ss *ScopedState) Prefix() string

Prefix returns the write prefix for this scope.

func (*ScopedState) Raw 

func (ss *ScopedState) Raw() *State

Raw returns the underlying State for orchestrator-level access.

func (*ScopedState) Session 

func (ss *ScopedState) Session() *ScopedState

Session returns a ScopedState that writes to the session namespace.

func (*ScopedState) Set 

func (ss *ScopedState) Set(key string, value any)

Set writes a value to this scope's namespace.

type Server 

type Server struct {
	// contains filtered or unexported fields
}

Server serves agents over HTTP with JSON request/response. Routes: GET /health, POST /{name}/run

func NewServer 

func NewServer(agents ...*Agent) *Server

NewServer creates an HTTP handler serving the given agents. Each agent is accessible at POST /{agent.Name}/run.

func (*Server) ServeHTTP 

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP implements http.Handler.

type Skill 

type Skill struct {
	Name         string // unique identifier (from SKILL.md frontmatter)
	Description  string // what this skill does (from SKILL.md frontmatter)
	Instructions string // markdown body (appended to agent system prompt)
	Tools        []Tool // skill-provided tools (merged with agent tools)
	Dir          string // source directory (empty for programmatic skills)
}

Skill packages domain-specific instructions and tools for an agent. Load skills from disk with LoadSkill or LoadSkills, or construct programmatically for testing.

A skill directory follows the standard SKILL.md convention: 

my-skill/
  SKILL.md          # name, description, instructions
  references/       # optional reference files (auto-generates tools)
    api-spec.md
    examples.txt

func LoadSkill 

func LoadSkill(dir string) (*Skill, error)

LoadSkill reads a skill from a directory containing a SKILL.md file. If the directory contains a references/ subdirectory, tools for listing and reading reference files are auto-generated.

The skill's Tools field is populated only with reference tools (if any). To add programmatic tools, append to the returned Skill's Tools field.

func LoadSkills 

func LoadSkills(parentDir string) ([]*Skill, error)

LoadSkills reads all skills from subdirectories of parentDir. Each subdirectory must contain a SKILL.md file. Subdirectories without SKILL.md are silently skipped. Returns an error only if parentDir cannot be read or a skill with SKILL.md fails to parse.

type State 

type State struct {
	// contains filtered or unexported fields
}

State is a thread-safe key-value store shared across agents, tools, and orchestrators. It replaces both the old StateFromContext map and GraphState.

State uses a flat map with scope prefixes by convention:

  • "session::" — visible to all agents in a session
  • "agent::<name>::" — scoped to a single agent's tools
  • "node::<id>::" — output of a graph node
  • "graph::" — graph-level metadata
  • no prefix — global

func NewState 

func NewState() *State

NewState creates an empty state.

func StateFromContext 

func StateFromContext(ctx context.Context) *State

StateFromContext retrieves the State from context. Returns nil if none.

func (*State) Delete 

func (s *State) Delete(key string)

Delete removes a key.

func (*State) Get 

func (s *State) Get(key string) (any, bool)

Get retrieves a value. Returns nil, false if not found.

func (*State) GetString 

func (s *State) GetString(key string) string

GetString retrieves a string value. Returns "" if not found or not a string.

func (*State) Keys 

func (s *State) Keys(prefix string) []string

Keys returns all keys matching an optional prefix. Pass "" for all keys.

func (*State) Set 

func (s *State) Set(key string, value any)

Set stores a value at the given key.

func (*State) Snapshot 

func (s *State) Snapshot() map[string]any

Snapshot returns a shallow copy of all data.

type StreamEvent 

type StreamEvent struct {
	Type    string         `json:"type"`
	Content string         `json:"content,omitempty"` // token text
	Agent   string         `json:"agent,omitempty"`   // current agent name
	Data    map[string]any `json:"data,omitempty"`    // event-specific payload
	Error   string         `json:"error,omitempty"`   // serializable error string
	Err     error          `json:"-"`                 // Go-side error (not serialized)
}

StreamEvent is an orchestrator-level streaming event for callers and UIs.

Agent events: "token", "tool_call.start", "tool_call.done", "done", "error" Swarm adds: "agent.start", "agent.done", "handoff" Graph adds: "node.start", "node.done", "edge"

type Streamer 

type Streamer interface {
	// Stream returns a channel that emits events as the LLM generates output.
	Stream(ctx context.Context, messages []Message, tools []map[string]any) <-chan Event
}

Streamer is an optional interface for streaming responses token by token.

type Swarm 

type Swarm struct {
	Name            string              // swarm identifier
	Agents          map[string]*Agent   // available agents by name
	Handoffs        map[string][]string // allowed transfers: agent -> targets (nil = all-to-all)
	EntryPoint      string              // starting agent (defaults to first agent)
	MaxHandoffs     int                 // loop limit; 0 means DefaultMaxHandoffs
	Observer        Observer            // receives events during execution (optional)
	CheckpointStore CheckpointStore     // enables durable execution (optional)
}

Swarm orchestrates multiple agents with dynamic handoffs. Each agent can transfer control to another agent mid-conversation using auto-generated transfer_to_<agent> tools.

All agents in a swarm share a single State, enabling structured data to flow across handoffs without serialization loss.

func (*Swarm) Resume 

func (s *Swarm) Resume(ctx context.Context, runID string) (*SwarmResult, error)

Resume loads the latest checkpoint for the given run and continues execution.

func (*Swarm) Run 

func (s *Swarm) Run(ctx context.Context, input string) (*SwarmResult, error)

Run executes the swarm starting from the entry point agent. A single State is shared across all agents, enabling cross-agent data flow through scoped state access.

func (*Swarm) Stream 

func (s *Swarm) Stream(ctx context.Context, input string) <-chan StreamEvent

Stream executes the swarm and streams events as they occur. It returns a channel that emits StreamEvent values. The channel is closed when the swarm finishes. Events include agent.start, agent.done, handoff, and all forwarded agent events (token, tool_call.start, tool_call.done).

type SwarmCheckpoint 

type SwarmCheckpoint struct {
	SwarmName     string      `json:"swarm_name"`
	OriginalInput string      `json:"original_input"`
	CurrentAgent  string      `json:"current_agent"`
	CurrentInput  string      `json:"current_input"`
	HandoffChain  []string    `json:"handoff_chain"`
	HandoffCount  int         `json:"handoff_count"`
	Steps         []SwarmStep `json:"steps"`
	PromptTokens  int         `json:"prompt_tokens"`
	OutputTokens  int         `json:"output_tokens"`
	ToolCalls     int         `json:"tool_calls"`
}

SwarmCheckpoint captures swarm loop state at a clean boundary.

type SwarmResult 

type SwarmResult struct {
	Output       string
	Steps        []SwarmStep
	HandoffChain []string
	PromptTokens int
	OutputTokens int
	ToolCalls    int
	DurationMS   int64
	Cost         float64
}

SwarmResult contains the outcome of a swarm run.

type SwarmStep 

type SwarmStep struct {
	Agent         string
	Input         string
	Output        string
	HandoffTo     string
	HandoffReason string
	PromptTokens  int
	OutputTokens  int
	ToolCalls     int
	DurationMS    int64
}

SwarmStep records one agent's execution in the swarm.

type Tool 

type Tool interface {
	Name() string
	Description() string
	Schema() map[string]any
	// Execute runs the tool. State provides scoped read/write access.
	Execute(ctx context.Context, state *ScopedState, args map[string]any) (ToolResult, error)
}

Tool is the interface for agent tools. Use NewTool for type-safe tools or SimpleTool for quick definitions.

func BashTool 

func BashTool() Tool

BashTool creates a tool that executes shell commands. Commands run via "sh -c" and inherit the context for cancellation. Output is truncated at 64KB.

func CodeTools 

func CodeTools() []Tool

CodeTools returns the standard set of coding tools: read, write, edit, bash, think.

func EditTool 

func EditTool() Tool

EditTool creates a tool that performs find-and-replace edits on files. The old_string must appear exactly once in the file (edits must be unambiguous).

func NewTool 

func NewTool[T any](name, description string, fn func(context.Context, *ScopedState, T) (ToolResult, error)) Tool

NewTool creates a type-safe tool from a function. The function must have the signature: 

func(context.Context, *ScopedState, T) (ToolResult, error)

where T is a struct with json tags defining the parameters.

Example: 

type WeatherArgs struct {
    City string `json:"city" desc:"City name" required:"true"`
}

tool := tantra.NewTool("get_weather", "Get weather for a city",
    func(ctx context.Context, state *tantra.ScopedState, args WeatherArgs) (tantra.ToolResult, error) {
        return tantra.SimpleResult(fmt.Sprintf("Weather in %s: Sunny", args.City)), nil
    })

func ReadTool 

func ReadTool() Tool

ReadTool creates a tool that reads text files.

func SimpleTool 

func SimpleTool(name, description string, params []Param, fn func(ctx context.Context, state *ScopedState, args map[string]any) (ToolResult, error)) Tool

SimpleTool creates a tool with explicit parameters (no generics).

func ThinkTool 

func ThinkTool() Tool

ThinkTool creates a tool that echoes the input back, allowing the LLM to reason step by step within the agent loop. The thought is returned as-is, keeping it in context for subsequent iterations.

func WriteTool 

func WriteTool() Tool

WriteTool creates a tool that creates or overwrites files. Parent directories are created automatically.

type ToolCall 

type ToolCall struct {
	ID        string         `json:"id"`
	Name      string         `json:"name"`
	Arguments map[string]any `json:"arguments"`
}

ToolCall represents a tool invocation requested by the LLM.

type ToolResult 

type ToolResult struct {
	// Output is what the LLM sees as the tool response.
	Output string

	// Next optionally names the next tool to call without an LLM round-trip.
	// If empty, normal LLM-driven flow continues.
	Next string

	// NextArgs are arguments for the chained tool call. Ignored if Next is empty.
	NextArgs map[string]any
}

ToolResult is the structured return from a tool execution. To mutate state, tools should call ScopedState.Set directly.

func ChainResult 

func ChainResult(output, nextTool string, nextArgs map[string]any) ToolResult

ChainResult creates a ToolResult that triggers a follow-up tool call.

func SimpleResult 

func SimpleResult(output string) ToolResult

SimpleResult creates a ToolResult with just an output string.

Source Files

View all Source files

Directories

Path Synopsis
autonomous command
Autonomous agent: uses CodeTools to reason, write scripts, and execute them.
 Autonomous agent: uses CodeTools to reason, write scripts, and execute them.
chain command
Tool chaining: deterministic tool-to-tool calls without LLM round-trips.
 Tool chaining: deterministic tool-to-tool calls without LLM round-trips.
checkpoint command
Checkpointing: save and resume agent execution.
 Checkpointing: save and resume agent execution.
graph command
Example: Content pipeline using Graph orchestration
 Example: Content pipeline using Graph orchestration
graph-router command
Example: Graph with Router node for dynamic routing Demonstrates tool usage within graph nodes.
 Example: Graph with Router node for dynamic routing Demonstrates tool usage within graph nodes.
mcp command
MCP integration: connect to external tool servers.
 MCP integration: connect to external tool servers.
multi-turn command
Multi-turn conversation: pass message history between runs.
 Multi-turn conversation: pass message history between runs.
multimodal command
Multimodal input: images, audio, and files alongside text.
 Multimodal input: images, audio, and files alongside text.
observe command
Observability: structured events from every decision point.
 Observability: structured events from every decision point.
real command
A real agent example using the OpenAI API.
 A real agent example using the OpenAI API.
serve command
Example: HTTP server serving multiple agents
 Example: HTTP server serving multiple agents
skills command
Skills: load domain-specific knowledge from SKILL.md directories.
 Skills: load domain-specific knowledge from SKILL.md directories.
state command
State sharing: scoped state across agents and tool calls.
 State sharing: scoped state across agents and tool calls.
stream command
Streaming: real-time token delivery from Agent, Swarm, and Graph.
 Streaming: real-time token delivery from Agent, Swarm, and Graph.
swarm command
A swarm example with triage, billing, and support agents.
 A swarm example with triage, billing, and support agents.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.