agentictools

Overview ¶

Package agentictools provides a tool executor processor component that routes tool calls to registered tool executors with filtering and timeout support.

Package agentictools provides the tool execution processor for the SemStreams agentic system.

Overview ¶

The agentic-tools processor executes tool calls from the agentic loop orchestrator. It receives ToolCall messages, dispatches them to registered tool executors, and publishes ToolResult messages back. The processor supports tool registration, allowlist filtering, per-execution timeouts, and concurrent execution.

This processor enables agents to interact with external systems (files, APIs, databases, etc.) through a well-defined tool interface.

Architecture ¶

The tools processor sits between the loop orchestrator and tool implementations:

┌───────────────┐     ┌────────────────┐     ┌──────────────────┐
│ agentic-loop  │────▶│ agentic-tools  │────▶│ Tool Executors   │
│               │     │ (this pkg)     │     │ (your code)      │
│               │◀────│                │◀────│                  │
└───────────────┘     └────────────────┘     └──────────────────┘
  tool.execute.*        Execute()           read_file, query_db,
  tool.result.*                             call_api, etc.

ToolExecutor Interface ¶

Tools are implemented by satisfying the ToolExecutor interface:

type ToolExecutor interface {
    Execute(ctx context.Context, call agentic.ToolCall) (agentic.ToolResult, error)
    ListTools() []agentic.ToolDefinition
}

Example implementation:

type FileReader struct{}

func (f *FileReader) Execute(ctx context.Context, call agentic.ToolCall) (agentic.ToolResult, error) {
    path, _ := call.Arguments["path"].(string)

    // Respect context cancellation
    select {
    case <-ctx.Done():
        return agentic.ToolResult{CallID: call.ID, Error: "cancelled"}, ctx.Err()
    default:
    }

    content, err := os.ReadFile(path)
    if err != nil {
        return agentic.ToolResult{
            CallID: call.ID,
            Error:  err.Error(),
        }, nil
    }

    return agentic.ToolResult{
        CallID:  call.ID,
        Content: string(content),
    }, nil
}

func (f *FileReader) ListTools() []agentic.ToolDefinition {
    return []agentic.ToolDefinition{
        {
            Name:        "read_file",
            Description: "Read the contents of a file",
            Parameters: map[string]any{
                "type": "object",
                "properties": map[string]any{
                    "path": map[string]any{
                        "type":        "string",
                        "description": "Path to the file",
                    },
                },
                "required": []string{"path"},
            },
        },
    }
}

Tool Registration ¶

Tools can be registered in two ways:

1. Global registration via init() (preferred for reusable tools):

func init() {
    agentictools.RegisterTool("my_tool", &MyToolExecutor{})
}

Global registration makes tools available to all agentic-tools components automatically. This pattern matches how components and rules are registered.

2. Per-component registration (for component-specific tools):

comp, _ := agentictools.NewComponent(rawConfig, deps)
toolsComp := comp.(*agentictools.Component)
err := toolsComp.RegisterToolExecutor(&FileReader{})

The processor extracts tool names from ListTools() for routing and validation.

ExecutorRegistry ¶

The ExecutorRegistry provides thread-safe tool management:

registry := NewExecutorRegistry()

// Register tools
registry.RegisterTool("read_file", &FileReader{})
registry.RegisterTool("query_db", &DatabaseQuerier{})

// Get executor by name
executor := registry.GetTool("read_file")

// List all available tools
tools := registry.ListTools()

// Execute a tool call
result, err := registry.Execute(ctx, toolCall)

The registry prevents duplicate registrations and returns descriptive errors for missing tools.

Tool Allowlist ¶

The processor supports allowlist filtering for security and control:

config := agentictools.Config{
    AllowedTools: []string{"read_file", "list_dir"},  // Only these allowed
    // ...
}

Behavior:

• Empty/nil AllowedTools: All registered tools are allowed
• Populated AllowedTools: Only listed tools can execute
• Blocked tools return an error result (not a Go error)

Example blocked response:

result := agentic.ToolResult{
    CallID: "call_001",
    Error:  "tool 'delete_file' is not allowed",
}

Timeout Handling ¶

Each tool execution runs with a configurable timeout:

config := agentictools.Config{
    Timeout: "60s",  // Per-tool execution timeout
    // ...
}

The timeout is enforced via context cancellation. Tool implementations should respect ctx.Done() for proper cancellation:

func (t *SlowTool) Execute(ctx context.Context, call agentic.ToolCall) (agentic.ToolResult, error) {
    select {
    case <-ctx.Done():
        return agentic.ToolResult{CallID: call.ID, Error: "execution cancelled"}, ctx.Err()
    case result := <-t.doWork(call):
        return result, nil
    }
}

Timeout errors are returned as ToolResult.Error, not as Go errors.

Quick Start ¶

Configure and start the processor:

config := agentictools.Config{
    StreamName:   "AGENT",
    AllowedTools: nil,  // Allow all
    Timeout:      "60s",
}

rawConfig, _ := json.Marshal(config)
comp, err := agentictools.NewComponent(rawConfig, deps)

// Register tools
toolsComp := comp.(*agentictools.Component)
toolsComp.RegisterToolExecutor(&FileReader{})
toolsComp.RegisterToolExecutor(&WebFetcher{})

// Start
lc := comp.(component.LifecycleComponent)
lc.Initialize()
lc.Start(ctx)

Configuration Reference ¶

Full configuration schema:

{
    "allowed_tools": ["string", ...],
    "timeout": "string (default: 60s)",
    "stream_name": "string (default: AGENT)",
    "consumer_name_suffix": "string (optional)",
    "ports": {
        "inputs": [...],
        "outputs": [...]
    }
}

Configuration fields:

• allowed_tools: List of tool names to allow (nil/empty = allow all)
• timeout: Per-tool execution timeout as duration string (default: "60s")
• stream_name: JetStream stream name for agentic messages (default: "AGENT")
• consumer_name_suffix: Optional suffix for JetStream consumer names (for testing)
• ports: Port configuration for inputs and outputs

Ports ¶

Input ports (JetStream consumers):

• tool.execute: Tool execution requests from agentic-loop (subject: tool.execute.>)

Output ports (JetStream publishers):

• tool.result: Tool execution results to agentic-loop (subject: tool.result.*)

Message Flow ¶

The processor handles each tool call through:

1. Receive ToolCall from tool.execute.>
2. Validate tool is in allowlist (if configured)
3. Look up executor in registry
4. Create timeout context
5. Execute tool with context
6. Publish ToolResult to tool.result.{call_id}
7. Acknowledge JetStream message

Error Handling ¶

Errors are categorized into two types:

**Tool execution errors** (returned in ToolResult.Error):

• Tool not found in registry
• Tool not in allowlist
• Tool execution failed
• Timeout exceeded

**System errors** (returned as Go error):

• JSON marshaling failures
• NATS publishing failures

Tool execution errors don't fail the loop - the agent can handle them:

if result.Error != "" {
    // Agent sees: "Error: file not found"
    // Agent can try alternative approach
}

Concurrent Execution ¶

The processor handles concurrent tool calls safely:

• ExecutorRegistry uses RWMutex for thread-safe access
• Each tool call gets its own goroutine
• Context cancellation propagates to all active executions

Multiple tools can execute in parallel when the loop sends concurrent calls:

// agentic-loop sends two tool calls
tool.execute.read_file  → executes
tool.execute.query_db   → executes concurrently

Built-in Tools ¶

The package does not include built-in tools - all tools must be registered by the application. This keeps the processor focused and allows full control over available capabilities.

Common tools to implement:

• File operations: read_file, write_file, list_dir
• Web operations: fetch_url, call_api
• Database operations: query, insert, update
• Graph operations: graph_query (query knowledge graph)

Thread Safety ¶

The Component is safe for concurrent use after Start() is called:

• ExecutorRegistry uses RWMutex for all operations
• Tool registration should complete before Start()
• Multiple tool calls can execute concurrently

Testing ¶

For testing, use mock executors and unique consumer names:

type MockExecutor struct {
    result agentic.ToolResult
}

func (m *MockExecutor) Execute(ctx context.Context, call agentic.ToolCall) (agentic.ToolResult, error) {
    m.result.CallID = call.ID
    return m.result, nil
}

func (m *MockExecutor) ListTools() []agentic.ToolDefinition {
    return []agentic.ToolDefinition{{Name: "mock_tool"}}
}

// In test
config := agentictools.Config{
    StreamName:         "AGENT",
    ConsumerNameSuffix: "test-" + t.Name(),
}

Limitations ¶

Current limitations:

• No tool versioning (single version per name)
• No tool dependencies or ordering
• No streaming tool output
• No built-in rate limiting per tool
• Timeout is global, not per-tool configurable

See Also ¶

Related packages:

• agentic: Shared types (ToolCall, ToolResult, ToolDefinition)
• processor/agentic-loop: Loop orchestration
• processor/agentic-model: LLM endpoint integration

JetStream Integration ¶

All messaging uses JetStream for durability. Tool call subjects require the AGENT stream to exist with subjects matching tool.execute.> and tool.result.>.

Consumer naming follows the pattern: agentic-tools-{subject-pattern}

Package agentictools provides Prometheus metrics for agentic-tools component.