# Contributing
Source: https://laddr.agnetlabs.com/community/contributing
How to contribute to the Laddr project - development workflow, code style, and areas for contribution
# Contributing to Laddr
Thank you for your interest in contributing to Laddr! This guide will help you get started.
***
## Getting Started
### 1. Fork the Repository
Fork the [Laddr repository](https://github.com/AgnetLabs/Laddr) on GitHub.
### 2. Clone Your Fork
```bash theme={null}
git clone https://github.com/your-username/Laddr.git
cd Laddr
```
### 3. Set Up Development Environment
```bash theme={null}
# Create virtual environment
python3 -m venv venv
source venv/bin/activate
# Install in development mode
pip install -e lib/laddr
# Install development dependencies
pip install -r requirements-dev.txt
```
***
## Development Workflow
### 1. Create a Branch
```bash theme={null}
git checkout -b feature/your-feature-name
```
### 2. Make Changes
Make your changes following the project's coding standards.
### 3. Test Your Changes
```bash theme={null}
# Run tests
pytest
# Run linter
ruff check .
# Run type checker
mypy lib/laddr
```
### 4. Commit Your Changes
```bash theme={null}
git add .
git commit -m "feat: Add new feature"
```
### 5. Push and Create Pull Request
```bash theme={null}
git push origin feature/your-feature-name
```
Then create a pull request on GitHub.
***
## Code Style
### Python Style
* Follow PEP 8
* Use type hints
* Write docstrings for all functions and classes
* Keep functions focused and small
### Commit Messages
Follow conventional commits:
* `feat:` - New feature
* `fix:` - Bug fix
* `docs:` - Documentation changes
* `refactor:` - Code refactoring
* `test:` - Test additions/changes
***
## Areas for Contribution
### Documentation
* Improve existing documentation
* Add examples
* Fix typos and errors
### Code
* Bug fixes
* New features
* Performance improvements
* Test coverage
### Examples
* Add example agents
* Create tutorial content
* Share use cases
***
## Questions?
If you have questions, please:
1. Check existing [issues on GitHub](https://github.com/AgnetLabs/Laddr/issues)
2. [Open a new issue](https://github.com/AgnetLabs/Laddr/issues/new) for discussion
3. Join our community discussions
***
## Next Steps
* [Getting Started](/getting-started/install) - Setup guide
* [Agent Configuration](/guides/agents/agent-config) - Learn about agents
# Filesystem Agent Example
Source: https://laddr.agnetlabs.com/examples/filesystem-agent
Complete example of a filesystem exploration agent using MCP - setup, usage, and customization
A complete example of a filesystem exploration agent that can navigate directories, read files, and analyze codebases.
***
## Overview
This agent uses the MCP filesystem server to:
* Navigate directories
* Read and analyze files
* Search for content
* Provide file information
***
## Setup
### Install MCP Filesystem Server
```bash theme={null}
# Ensure Node.js is installed
node --version
# The MCP server will be installed automatically via npx
```
### Create Agent
```python theme={null}
# agents/filesystem_agent.py
import asyncio
from laddr import Agent, WorkerRunner
from laddr.llms import gemini
from laddr.core.mcp_tools import MCPToolProvider
async def main():
# Initialize MCP provider for filesystem
mcp = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-filesystem /path/to/project",
transport="stdio",
server_name="filesystem"
)
# Create agent
agent = Agent(
name="filesystem_assistant",
role="Filesystem Assistant",
goal="Help users explore and analyze files and directories",
backstory="""You are a helpful filesystem assistant. You can:
- Navigate directories
- Read and analyze files
- Search for content
- Provide file information""",
llm=gemini(model="gemini-2.5-flash", temperature=0.3),
tools=[mcp],
instructions="""
When helping users:
1. Use filesystem tools to explore directories
2. Read relevant files to answer questions
3. Provide clear, organized information
4. Be concise and focus on what the user asked
"""
)
# Run as worker
runner = WorkerRunner(agent=agent)
print("Starting filesystem assistant...")
await runner.start()
if __name__ == "__main__":
asyncio.run(main())
```
***
## Usage
### Run the Agent
```bash theme={null}
python agents/filesystem_agent.py
```
### Submit Tasks
```bash theme={null}
# List files in a directory
laddr prompt run filesystem_assistant --input query="List all Python files in the src directory"
# Read a specific file
laddr prompt run filesystem_assistant --input query="Read the contents of main.py"
# Search for content
laddr prompt run filesystem_assistant --input query="Find all files containing 'async def'"
```
***
## Example Interactions
### Directory Navigation
**User:** "Show me the structure of the docs directory"
**Agent:** Uses `list_directory` tool to explore and presents a tree structure.
### File Analysis
**User:** "Analyze the main.py file and explain what it does"
**Agent:** Uses `read_file` to read the file, then analyzes and explains the code.
### Content Search
**User:** "Find all files that import 'laddr'"
**Agent:** Uses search tools to find matching files and presents results.
***
## Customization
### Add More Tools
```python theme={null}
from laddr.core.mcp_tools import MCPToolProvider, MultiMCPToolProvider
# Filesystem MCP
fs_mcp = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-filesystem .",
transport="stdio",
server_name="filesystem"
)
# Git MCP (if needed)
git_mcp = MCPToolProvider(
command="uvx mcp-server-git",
transport="stdio",
server_name="git"
)
# Combine
multi_mcp = MultiMCPToolProvider([fs_mcp, git_mcp])
agent = Agent(
name="developer_assistant",
tools=[multi_mcp],
# ... other config
)
```
***
## Next Steps
* [MCP Integration](/guides/tools/mcp-integration) - Learn about MCP
* [MCP Examples](/examples/mcp-examples) - More MCP examples
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
# MCP Examples
Source: https://laddr.agnetlabs.com/examples/mcp-examples
Collection of MCP integration examples for Laddr agents - filesystem, Git, PostgreSQL, and HTTP servers
Collection of examples showing how to integrate various MCP servers with Laddr agents.
***
## Filesystem MCP
Explore and analyze files and directories.
```python theme={null}
from laddr import Agent
from laddr.core.mcp_tools import MCPToolProvider
mcp = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-filesystem /path/to/project",
transport="stdio",
server_name="filesystem"
)
agent = Agent(
name="filesystem_agent",
tools=[mcp],
# ... config
)
```
***
## Git MCP
Perform Git operations on repositories.
```python theme={null}
from laddr import Agent
from laddr.core.mcp_tools import MCPToolProvider
mcp = MCPToolProvider(
command="uvx mcp-server-git",
transport="stdio",
server_name="git"
)
agent = Agent(
name="git_agent",
tools=[mcp],
# ... config
)
```
***
## PostgreSQL MCP
Query PostgreSQL databases.
```python theme={null}
from laddr import Agent
from laddr.core.mcp_tools import MCPToolProvider
mcp = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-postgres postgresql://user:pass@localhost:5432/db",
transport="stdio",
server_name="postgres"
)
agent = Agent(
name="database_agent",
tools=[mcp],
# ... config
)
```
***
## Multiple MCP Servers
Combine multiple MCP servers.
```python theme={null}
from laddr import Agent
from laddr.core.mcp_tools import MCPToolProvider, MultiMCPToolProvider
# Filesystem
fs_mcp = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-filesystem .",
transport="stdio",
server_name="filesystem"
)
# Git
git_mcp = MCPToolProvider(
command="uvx mcp-server-git",
transport="stdio",
server_name="git"
)
# Combine
multi_mcp = MultiMCPToolProvider([fs_mcp, git_mcp])
agent = Agent(
name="developer_assistant",
tools=[multi_mcp],
# ... config
)
```
***
## HTTP MCP Server
Connect to remote HTTP-based MCP servers.
```python theme={null}
from laddr import Agent
from laddr.core.mcp_tools import MCPToolProvider
mcp = MCPToolProvider(
url="https://mcp-server.example.com",
transport="http",
api_key="your-api-key"
)
agent = Agent(
name="remote_agent",
tools=[mcp],
# ... config
)
```
***
## Next Steps
* [MCP Integration](/guides/tools/mcp-integration) - Complete MCP guide
* [Filesystem Agent](/examples/filesystem-agent) - Complete filesystem example
* [Tool Configuration](/guides/tools/tool-config) - Create custom tools
# Your First Agent
Source: https://laddr.agnetlabs.com/getting-started/first-agent
Build your first Laddr agent from scratch - step-by-step tutorial to create a web research agent
Learn to create a custom agent that can search the web and summarize information.
***
## Overview
In this tutorial, you'll build a **researcher agent** that:
1. Takes a research query
2. Searches the web for information
3. Summarizes the findings
***
## Step 1: Create the Agent File
Create `agents/researcher.py`:
```python theme={null}
from __future__ import annotations
import asyncio
from laddr import Agent, WorkerRunner
from laddr.llms import gemini
from tools.web_tools import web_search
# Define the researcher agent
researcher = Agent(
name="researcher",
role="Web Research Specialist",
goal="Search the web and summarize findings",
backstory="""You are a research-focused agent who finds accurate
data online and presents it clearly.""",
llm=gemini(model="gemini-2.5-flash", temperature=0.0),
tools=[web_search],
max_retries=1,
max_iterations=3,
max_tool_calls=2,
timeout=45,
trace_enabled=True,
)
async def main():
runner = WorkerRunner(agent=researcher)
print("Starting researcher worker...")
await runner.start()
if __name__ == "__main__":
asyncio.run(main())
```
***
## Step 2: Create a Tool
Create `tools/web_tools.py`:
```python theme={null}
from laddr import tool
import os
import requests
@tool(
name="web_search",
description="Search the web for information on a given topic",
parameters={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query"
}
},
"required": ["query"]
}
)
def web_search(query: str) -> dict:
"""Search the web using Serper API."""
api_key = os.getenv("SERPER_API_KEY")
if not api_key:
return {"error": "SERPER_API_KEY not set"}
url = "https://google.serper.dev/search"
headers = {
"X-API-KEY": api_key,
"Content-Type": "application/json"
}
payload = {"q": query}
try:
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
data = response.json()
# Extract top results
results = []
if "organic" in data:
for item in data["organic"][:5]:
results.append({
"title": item.get("title", ""),
"snippet": item.get("snippet", ""),
"link": item.get("link", "")
})
return {
"status": "success",
"results": results,
"total": len(results)
}
except Exception as e:
return {"status": "error", "error": str(e)}
```
You can use any search API. Serper is used here as an example. You can also use DuckDuckGo, Bing, or other providers.
***
## Step 3: Set Environment Variables
Add to your `.env` file:
```bash theme={null}
GEMINI_API_KEY=your_gemini_api_key
SERPER_API_KEY=your_serper_api_key
```
***
## Step 4: Run the Agent
### Option A: Local Run (Fastest)
```bash theme={null}
laddr run-local researcher --input '{"query": "What is Laddr?"}'
```
### Option B: As a Worker
Start the worker:
```bash theme={null}
python agents/researcher.py
```
In another terminal, submit a task:
```bash theme={null}
laddr prompt run researcher --input query="Latest AI trends"
```
***
## Step 5: Understand the Output
You should see output like:
```
Starting researcher worker...
Processing task: {'query': 'What is Laddr?'}
Using tool: web_search
Tool result: {'status': 'success', 'results': [...], 'total': 5}
Generating summary...
Agent response: Based on my research, Laddr is an open-source multi-agent framework...
Task completed successfully
```
***
## Understanding Agent Configuration
Let's break down the agent configuration:
### Core Parameters
```python theme={null}
Agent(
name="researcher", # Unique identifier
role="Web Research Specialist", # Agent's role
goal="Search the web...", # What the agent does
backstory="...", # Context for behavior
)
```
### LLM Configuration
```python theme={null}
llm=gemini(
model="gemini-2.5-flash", # Model name
temperature=0.0 # Lower = more deterministic
)
```
### Execution Control
```python theme={null}
max_retries=1, # Retry failed tasks once
max_iterations=3, # Max reasoning loops
max_tool_calls=2, # Max tool calls per iteration
timeout=45, # Timeout in seconds
```
***
## Adding More Tools
Add additional tools to your agent:
```python theme={null}
from tools.web_tools import web_search, scrape_url, extract_links
researcher = Agent(
name="researcher",
# ... other config
tools=[web_search, scrape_url, extract_links], # Multiple tools
)
```
***
## Advanced: Custom Instructions
Add custom instructions to guide agent behavior:
```python theme={null}
researcher = Agent(
name="researcher",
# ... other config
instructions="""
When researching:
1. Use web_search to find relevant information
2. Focus on recent and authoritative sources
3. Summarize findings concisely
4. Cite sources when possible
""",
)
```
***
## Testing Your Agent
Test with different queries:
```bash theme={null}
# Simple query
laddr run-local researcher --input '{"query": "Python async programming"}'
# Complex query
laddr run-local researcher --input '{"query": "Compare Laddr vs LangGraph"}'
```
***
## Next Steps
Learn all agent configuration options
Create more sophisticated tools
Best practices for agent design
Connect to MCP servers
***
## Troubleshooting
### Agent Not Found
Ensure your agent file is in the `agents/` directory and properly imported.
### Tool Errors
Check that:
* Tool is properly decorated with `@tool`
* Parameters match the schema
* API keys are set in `.env`
### LLM Errors
Verify:
* API key is valid
* Model name is correct
* Network connectivity
***
## Next Steps
* [Agent Configuration](/guides/agents/agent-config) - Complete configuration reference
* [Tool Development](/guides/tools/tool-config) - Build custom tools
* [Agent Authoring](/guides/agents/authoring) - Design patterns
* [Playground](/getting-started/playground) - Use the dashboard
# Installation
Source: https://laddr.agnetlabs.com/getting-started/install
Install and set up Laddr on your system - complete setup guide with Docker, Python, and environment configuration
Get Laddr installed and configured on your system.
***
## Prerequisites
* **Python 3.9+** - Required for Laddr
* **pip** - Python package manager
* **Docker** (optional but recommended) - For full stack with dashboard
* **Node.js** (optional) - For MCP servers
***
## Python Virtual Environment
Create a virtual environment to isolate dependencies:
```bash theme={null}
python3 -m venv venv
```
Activate the virtual environment:
```bash theme={null}
# On macOS/Linux
source venv/bin/activate
# On Windows (Git Bash or CMD)
.\\venv\\Scripts\\activate
```
Always activate your virtual environment before installing or using Laddr.
***
## Install Laddr
### Standard Installation
```bash theme={null}
pip install laddr
```
This installs:
* Laddr CLI
* Core framework
* API server
* Worker runtime
### Development Installation
To develop against the Laddr repository:
```bash theme={null}
pip install -e lib/laddr
```
### Verify Installation
```bash theme={null}
laddr --version
laddr --help
```
***
## Docker Setup
For the full experience including the web UI and isolated environments, install Docker:
1. Download from [Docker Desktop](https://www.docker.com/products/docker-desktop)
2. Install and start Docker
3. Verify installation:
```bash theme={null}
docker --version
docker compose version
```
***
## Create a New Project
Initialize a new Laddr project:
```bash theme={null}
laddr init my_agent_system
cd my_agent_system
```
This creates:
```
my_agent_system/
├── agents/ # Agent definitions
├── workers/ # Worker scripts
├── tools/ # Custom tools
├── Dockerfile # Docker configuration
├── docker-compose.yml
├── main.py # Runner script
└── .env.example # Environment template
```
***
## Configure Environment
Copy the example environment file:
```bash theme={null}
cp .env.example .env
```
Edit `.env` with your configuration:
```bash theme={null}
# .env
# LLM Configuration
GEMINI_API_KEY=your_gemini_api_key
# or
OPENAI_API_KEY=sk-...
# or
ANTHROPIC_API_KEY=sk-ant-...
# Optional: Tool API keys
SERPER_API_KEY=your_serper_api_key
# Queue Backend (default: memory)
QUEUE_BACKEND=memory # or redis, kafka
# Database (default: sqlite)
DB_BACKEND=sqlite # or postgresql
DATABASE_URL=sqlite:///./laddr.db
# Storage (default: local)
STORAGE_BACKEND=local # or minio, s3
```
Never commit your `.env` file to version control. It contains sensitive API keys.
***
## Run the Stack
### Option 1: Local Development (No Docker)
For quick testing without Docker:
```bash theme={null}
# Set environment
export QUEUE_BACKEND=memory
export DB_BACKEND=sqlite
# Run agent locally
laddr run-local researcher --input '{"query": "test"}'
```
### Option 2: Full Stack (Docker)
Start all services with Docker:
```bash theme={null}
laddr run dev -d
# or
docker compose up -d
```
This starts:
* **PostgreSQL** - Database (port 5432)
* **Redis** - Message queue (port 6379)
* **MinIO** - Object storage (port 9000)
* **API Server** - REST API (port 8000)
* **Dashboard** - Web UI (port 5173)
* **Workers** - Agent workers
Access the dashboard at `http://localhost:5173`
***
## Verify Installation
### Check Services
```bash theme={null}
laddr ps
```
### Run Diagnostics
```bash theme={null}
laddr check
```
This validates:
* Database connectivity
* Queue backend connection
* Storage backend access
* LLM API connectivity
* Agent configuration
### Test with a Simple Agent
```bash theme={null}
# Add a test agent
laddr add agent tester \\
--role "Test Agent" \\
--goal "Test the system" \\
--llm-model gemini-2.5-flash
# Run it
laddr run-local tester --input '{"message": "Hello"}'
```
***
## Troubleshooting
### Command Not Found
If `laddr` command is not found:
```bash theme={null}
# Check if installed
pip show laddr
# Reinstall if needed
pip install --upgrade laddr
# Or use Python module
python -m laddr --version
```
### Docker Issues
If Docker commands fail:
```bash theme={null}
# Check Docker is running
docker ps
# Check Docker Compose
docker compose version
# Restart Docker Desktop if needed
```
### Port Conflicts
If ports are already in use:
```bash theme={null}
# Check what's using the port
lsof -i :8000 # macOS/Linux
netstat -ano | findstr :8000 # Windows
# Change ports in docker-compose.yml
```
### API Key Errors
Ensure your API keys are valid:
```bash theme={null}
# Test API key
curl https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key=YOUR_KEY
```
***
## Next Steps
* [First Agent](/getting-started/first-agent) - Build your first agent
* [Playground](/getting-started/playground) - Explore the dashboard
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
* [Local Runtime](/guides/local-runtime) - Local development guide
# Playground & Dashboard
Source: https://laddr.agnetlabs.com/getting-started/playground
Use the Laddr dashboard to monitor and interact with agents - real-time traces, job history, and system metrics
The Laddr dashboard provides a web-based interface for monitoring agents, viewing traces, and interacting with your multi-agent system.
***
## Starting the Dashboard
Start the full Laddr stack:
```bash theme={null}
laddr run dev -d
# or
docker compose up -d
```
The dashboard is available at:
**[http://localhost:5173](http://localhost:5173)**
The dashboard automatically connects to the API server at `http://localhost:8000`.
***
## Dashboard Features
### Agent Overview
View all registered agents and their status:
* **Agent List** - All agents in your system
* **Status Indicators** - Running, idle, or error states
* **Worker Count** - Number of active workers per agent
### Real-Time Traces
Monitor agent execution in real-time:
* **Execution Timeline** - See each step as it happens
* **Tool Calls** - View tool invocations and results
* **LLM Interactions** - See prompts and responses
* **Token Usage** - Track LLM costs
### Job History
Browse past executions:
* **Job List** - All completed jobs
* **Search & Filter** - Find specific jobs
* **Job Details** - View full execution trace
* **Replay Jobs** - Re-run previous jobs
### System Metrics
Monitor system health:
* **Queue Status** - Message queue health
* **Database Status** - Connection status
* **Storage Status** - Storage backend health
* **Worker Metrics** - Active workers and throughput
***
## Using the Playground
### Submit a Job
1. Navigate to the **Playground** tab
2. Select an agent from the dropdown
3. Enter input JSON:
```json theme={null}
{
"query": "What is Laddr?",
"max_results": 5
}
```
4. Click **Submit**
5. Watch the execution in real-time
### View Traces
1. Click on any job in the **Jobs** tab
2. View the execution trace:
* **Timeline** - Chronological view
* **Tree View** - Hierarchical structure
* **Raw JSON** - Complete trace data
### Replay Jobs
1. Find a job in the **Jobs** tab
2. Click **Replay**
3. Choose:
* **Return stored result** - Get cached result
* **Re-execute** - Run the job again
***
## API Integration
The dashboard uses the REST API. You can also use the API directly:
### Submit via API
```bash theme={null}
curl -X POST http://localhost:8000/api/prompts \\
-H "Content-Type: application/json" \\
-d '{
"prompt_name": "researcher",
"inputs": {"query": "What is Laddr?"}
}'
```
### Get Job Status
```bash theme={null}
curl http://localhost:8000/api/prompts/{prompt_id}
```
### List Jobs
```bash theme={null}
curl "http://localhost:8000/api/prompts?limit=50"
```
***
## Monitoring Best Practices
### Watch for Errors
Monitor the dashboard for:
* **Failed Jobs** - Check error messages
* **Timeout Errors** - Increase timeout or optimize agent
* **Tool Errors** - Verify tool configuration
* **LLM Errors** - Check API keys and quotas
### Track Performance
Use metrics to identify:
* **Slow Agents** - High execution times
* **High Token Usage** - Optimize prompts
* **Queue Backlog** - Scale workers
* **Storage Usage** - Monitor artifact storage
### Debug Issues
Use traces to debug:
1. **View Full Trace** - See every step
2. **Check Tool Results** - Verify tool outputs
3. **Review LLM Calls** - Inspect prompts/responses
4. **Compare Jobs** - Find patterns in failures
***
## Keyboard Shortcuts
* **`/`** - Focus search
* **`?`** - Show shortcuts
* **`Esc`** - Close modals
* **`Ctrl+K`** - Quick actions
***
## Troubleshooting
### Dashboard Not Loading
```bash theme={null}
# Check if services are running
laddr ps
# Check dashboard logs
docker compose logs dashboard
# Restart services
docker compose restart dashboard
```
### API Connection Errors
Verify API server is running:
```bash theme={null}
curl http://localhost:8000/api/health
```
### No Jobs Showing
Ensure workers are running:
```bash theme={null}
laddr ps
# Start workers if needed
docker compose up -d researcher_worker coordinator_worker
```
***
## Next Steps
* [API Reference](/reference/api) - Complete API documentation
* [Scaling & Operations](/guides/scaling-and-ops) - Production deployment
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
* [Local Runtime](/guides/local-runtime) - Local development
# Agent Configuration
Source: https://laddr.agnetlabs.com/guides/agents/agent-config
Complete reference for configuring Laddr agents - all parameters, options, and best practices
Complete guide to configuring Laddr agents with all available parameters and options.
***
## Agent Definition Structure
Basic agent definition:
```python theme={null}
from __future__ import annotations
import asyncio
import os
from dotenv import load_dotenv
from laddr import Agent, WorkerRunner
from laddr.llms import openai
from tools.web_tools import web_search, scrape_url, extract_links
load_dotenv()
TOOLS = [web_search, scrape_url, extract_links]
researcher = Agent(
name="researcher",
role="Web Research Specialist",
goal="Search the web and summarize findings",
backstory="You are a research-focused agent who finds accurate data online.",
llm=openai(model="gpt-4o-mini", temperature=0.0),
tools=TOOLS,
max_retries=1,
max_iterations=3,
max_tool_calls=2,
timeout=45,
trace_enabled=True,
)
async def main():
runner = WorkerRunner(agent=researcher)
print("Starting researcher worker...")
await runner.start()
if __name__ == "__main__":
asyncio.run(main())
```
***
## Core Agent Parameters
### Required Parameters
| Parameter | Type | Description |
| ----------- | ----- | ------------------------------------------ |
| `name` | `str` | Unique agent identifier (used in routing) |
| `role` | `str` | Agent's role description (for LLM context) |
| `goal` | `str` | Primary objective or task |
| `backstory` | `str` | Context that shapes agent behavior |
| `llm` | `LLM` | LLM configuration object |
### Optional Parameters
| Parameter | Type | Default | Description |
| ---------------- | ------ | ------- | -------------------------------------- |
| `tools` | `list` | `[]` | List of tool functions |
| `max_retries` | `int` | `1` | Number of retries for failed tasks |
| `max_iterations` | `int` | `3` | Maximum reasoning loops |
| `max_tool_calls` | `int` | `10` | Maximum tool calls per iteration |
| `timeout` | `int` | `300` | Task timeout in seconds |
| `trace_enabled` | `bool` | `True` | Enable execution tracing |
| `instructions` | `str` | `None` | Custom instructions for agent behavior |
| `is_coordinator` | `bool` | `False` | Enable delegation tools |
***
## LLM Configuration
### OpenAI
```python theme={null}
from laddr.llms import openai
agent = Agent(
name="researcher",
llm=openai(
model="gpt-4o-mini",
temperature=0.0,
max_tokens=2000
),
# ... other config
)
```
### Gemini
```python theme={null}
from laddr.llms import gemini
agent = Agent(
name="researcher",
llm=gemini(
model="gemini-2.5-flash",
temperature=0.0
),
# ... other config
)
```
### Anthropic Claude
```python theme={null}
from laddr.llms import anthropic
agent = Agent(
name="researcher",
llm=anthropic(
model="claude-3-5-sonnet-20241022",
temperature=0.0
),
# ... other config
)
```
### Ollama (Local)
```python theme={null}
from laddr.llms import ollama
agent = Agent(
name="researcher",
llm=ollama(
model="llama3.2:latest",
base_url="http://localhost:11434"
),
# ... other config
)
```
***
## Environment Configuration
### Queue and Database
```bash theme={null}
# .env
# Queue Backend
QUEUE_BACKEND=redis # or kafka, memory
REDIS_URL=redis://localhost:6379
KAFKA_BOOTSTRAP=kafka:9092
# Database
DB_BACKEND=sqlite # or postgresql
DATABASE_URL=sqlite:///./laddr.db
# or
DATABASE_URL=postgresql://user:pass@localhost:5432/laddr
```
### LLM Configuration
```bash theme={null}
# .env
# Global LLM settings
LLM_PROVIDER=openai
LLM_MODEL=gpt-4o-mini
LLM_TEMPERATURE=0.0
# API Keys
OPENAI_API_KEY=sk-...
GEMINI_API_KEY=your_gemini_api_key
ANTHROPIC_API_KEY=sk-ant-...
# Per-agent model overrides
RESEARCHER_MODEL=gpt-4o-mini
WRITER_MODEL=claude-3-5-sonnet-20241022
COORDINATOR_MODEL=gemini-2.5-flash
```
### Tool API Keys
```bash theme={null}
# .env
SERPER_API_KEY=your_serper_api_key
WEATHER_API_KEY=your_weather_api_key
# ... other tool keys
```
***
## Execution Control
### Retry Configuration
```python theme={null}
agent = Agent(
name="researcher",
max_retries=3, # Retry up to 3 times on failure
# ... other config
)
```
Set `max_retries` based on task criticality. Use 1-2 for non-critical tasks, 3-5 for important tasks.
### Iteration Limits
```python theme={null}
agent = Agent(
name="researcher",
max_iterations=5, # Allow up to 5 reasoning loops
max_tool_calls=10, # Max 10 tool calls per iteration
# ... other config
)
```
Too many iterations can lead to high costs and slow execution. Start with 3-5 iterations.
### Timeout Configuration
```python theme={null}
agent = Agent(
name="researcher",
timeout=60, # 60 second timeout
# ... other config
)
```
***
## Tool Configuration
### Registering Tools
```python theme={null}
from laddr import tool
@tool(
name="web_search",
description="Search the web",
parameters={
"type": "object",
"properties": {
"query": {"type": "string"}
}
}
)
def web_search(query: str):
# Tool implementation
pass
agent = Agent(
name="researcher",
tools=[web_search], # Register tool
# ... other config
)
```
### Multiple Tools
```python theme={null}
agent = Agent(
name="researcher",
tools=[
web_search,
scrape_url,
extract_links,
summarize_text
],
# ... other config
)
```
***
## Tracing Configuration
### Enable Tracing
```python theme={null}
agent = Agent(
name="researcher",
trace_enabled=True, # Enable execution traces
# ... other config
)
```
Traces include:
* Task start/complete events
* LLM calls (prompts, responses, tokens)
* Tool calls (inputs, outputs, errors)
* Autonomous thinking steps
### View Traces
```bash theme={null}
# Via dashboard
# http://localhost:5173
# Via API
curl http://localhost:8000/api/jobs/{job_id}
# Via CLI
laddr logs researcher --follow
```
***
## Worker Runtime
### Running as Worker
```python theme={null}
from laddr import Agent, WorkerRunner
researcher = Agent(
name="researcher",
# ... config
)
async def main():
runner = WorkerRunner(agent=researcher)
await runner.start()
if __name__ == "__main__":
import asyncio
asyncio.run(main())
```
### Local Execution
```bash theme={null}
# Run locally without worker
laddr run-local researcher --input '{"query": "test"}'
```
***
## Coordinator Agents
Enable delegation capabilities:
```python theme={null}
coordinator = Agent(
name="coordinator",
role="Task Coordinator",
goal="Coordinate multi-agent workflows",
is_coordinator=True, # Enables delegation tools
# ... other config
)
```
Coordinator agents can:
* Delegate tasks to other agents
* Run parallel delegations
* Coordinate multi-agent workflows
See [System Tools](/guides/system-tools/custom-system-tools) for more details.
***
## Custom Instructions
Add behavior guidance:
```python theme={null}
agent = Agent(
name="researcher",
instructions="""
When researching:
1. Use web_search to find information
2. Prioritize recent sources (last 2 years)
3. Cite sources in your response
4. Summarize findings in 2-3 paragraphs
""",
# ... other config
)
```
***
## Monitoring & Debugging
### View Logs
```bash theme={null}
# Follow logs in real-time
laddr logs researcher --follow
# Show last 100 lines
laddr logs researcher --tail 100
```
### Check Status
```bash theme={null}
# Show all services
laddr ps
# Run diagnostics
laddr check
```
### Dashboard
Access the dashboard at `http://localhost:5173` to:
* View real-time traces
* Monitor token usage
* Check system metrics
* Review job history
***
## Troubleshooting
### Agent Not Found
Ensure agent is properly registered:
```bash theme={null}
# Check agent file exists
ls agents/researcher.py
# Verify import
python -c "from agents.researcher import researcher; print(researcher.name)"
```
### LLM Errors
Check API keys and model names:
```bash theme={null}
# Verify API key
echo $OPENAI_API_KEY
# Test API connection
curl https://api.openai.com/v1/models \\
-H "Authorization: Bearer $OPENAI_API_KEY"
```
### Tool Errors
Verify tool configuration:
```bash theme={null}
# Check tool is registered
python -c "from agents.researcher import researcher; print(researcher.tools)"
# Verify tool decorator
grep -n "@tool" tools/web_tools.py
```
***
## Best Practices
### 1. Use Descriptive Names
```python theme={null}
# ✅ Good
researcher = Agent(name="researcher", ...)
data_analyzer = Agent(name="data_analyzer", ...)
# ❌ Bad
agent1 = Agent(name="agent1", ...)
a = Agent(name="a", ...)
```
### 2. Set Appropriate Timeouts
```python theme={null}
# Quick tasks
fast_agent = Agent(timeout=30, ...)
# Complex tasks
complex_agent = Agent(timeout=300, ...)
```
### 3. Limit Tool Count
```python theme={null}
# ✅ Good - 3-5 focused tools
agent = Agent(tools=[tool1, tool2, tool3], ...)
# ❌ Bad - Too many tools
agent = Agent(tools=[tool1, ..., tool20], ...)
```
### 4. Use Environment Variables
```python theme={null}
# ✅ Good - Configurable
llm=openai(model=os.getenv("LLM_MODEL", "gpt-4o-mini"))
# ❌ Bad - Hardcoded
llm=openai(model="gpt-4o-mini")
```
***
## Next Steps
* [Agent Authoring](/guides/agents/authoring) - Design patterns and best practices
* [Tool Configuration](/guides/tools/tool-config) - Create custom tools
* [System Tools](/guides/system-tools/custom-system-tools) - Customize delegation
* [Scaling & Operations](/guides/scaling-and-ops) - Production deployment
# Agent Authoring
Source: https://laddr.agnetlabs.com/guides/agents/authoring
Best practices and patterns for designing effective agents - design principles, patterns, and common pitfalls
Learn best practices for designing and authoring effective Laddr agents.
***
## Agent Design Principles
### Single Responsibility
Each agent should have a clear, focused purpose:
```python theme={null}
# ✅ Good - Focused agent
researcher = Agent(
name="researcher",
role="Web Research Specialist",
goal="Search the web and find accurate information",
# ...
)
# ❌ Bad - Too broad
general_agent = Agent(
name="general_agent",
role="Does everything",
goal="Handle all tasks",
# ...
)
```
### Clear Role Definition
Define the agent's role clearly:
```python theme={null}
researcher = Agent(
name="researcher",
role="Web Research Specialist", # What the agent is
goal="Find and summarize information from the web", # What it does
backstory="""You are an expert researcher with years of experience
finding accurate information online. You prioritize recent, authoritative
sources and present findings clearly.""", # How it behaves
)
```
### Appropriate Tools
Give agents only the tools they need:
```python theme={null}
# ✅ Good - Relevant tools
researcher = Agent(
name="researcher",
tools=[web_search, scrape_url, extract_links], # All research-related
)
# ❌ Bad - Unnecessary tools
researcher = Agent(
name="researcher",
tools=[web_search, calculator, weather_api], # Calculator not needed
)
```
***
## Agent Patterns
### Coordinator Pattern
A coordinator agent delegates tasks to specialized workers:
```python theme={null}
coordinator = Agent(
name="coordinator",
role="Task Coordinator",
goal="Break down complex tasks and delegate to specialists",
backstory="""You are an experienced project manager who breaks down
complex tasks into smaller, manageable pieces and assigns them to
the right specialists.""",
instructions="""
When given a task:
1. Analyze the requirements
2. Break it into subtasks
3. Delegate to appropriate agents
4. Synthesize results
""",
is_coordinator=True, # Enables delegation tools
)
```
### Worker Pattern
Specialized workers that handle specific tasks:
```python theme={null}
researcher = Agent(
name="researcher",
role="Research Specialist",
goal="Conduct thorough research on given topics",
# Focused on one task type
)
writer = Agent(
name="writer",
role="Content Writer",
goal="Write clear, engaging content",
# Focused on writing
)
```
### Pipeline Pattern
Agents in a sequential pipeline:
```python theme={null}
# Agent 1: Research
researcher = Agent(name="researcher", ...)
# Agent 2: Analyze
analyzer = Agent(name="analyzer", ...)
# Agent 3: Write
writer = Agent(name="writer", ...)
# Usage: researcher → analyzer → writer
```
***
## Instructions Design
### Clear Instructions
Write specific, actionable instructions:
```python theme={null}
agent = Agent(
name="analyzer",
instructions="""
When analyzing data:
1. First, identify the data type and structure
2. Look for patterns, trends, and anomalies
3. Calculate relevant statistics
4. Summarize findings in 2-3 bullet points
5. Highlight any concerns or recommendations
""",
)
```
### Avoid Ambiguity
Be explicit about expected behavior:
```python theme={null}
# ✅ Good - Clear expectations
agent = Agent(
instructions="""
Always cite sources when providing information.
Use the format: [Source Name](URL)
""",
)
# ❌ Bad - Vague
agent = Agent(
instructions="Be helpful and accurate",
)
```
### Include Examples
Provide examples in instructions:
```python theme={null}
agent = Agent(
instructions="""
Format responses as JSON with these fields:
- summary: Brief summary (max 100 words)
- key_points: Array of 3-5 key points
- sources: Array of source URLs
Example:
{
"summary": "...",
"key_points": ["...", "..."],
"sources": ["https://...", "https://..."]
}
""",
)
```
***
## Tool Selection
### Match Tools to Role
Select tools that align with the agent's purpose:
```python theme={null}
# Research agent needs web tools
researcher = Agent(
name="researcher",
tools=[web_search, scrape_url, extract_links],
)
# Data analyst needs data tools
analyst = Agent(
name="analyst",
tools=[read_csv, calculate_stats, generate_chart],
)
```
### Limit Tool Count
Too many tools can confuse the agent:
```python theme={null}
# ✅ Good - 3-5 focused tools
agent = Agent(
tools=[tool1, tool2, tool3], # Manageable set
)
# ❌ Bad - Too many tools
agent = Agent(
tools=[tool1, tool2, ..., tool20], # Overwhelming
)
```
***
## Error Handling
### Set Appropriate Retries
```python theme={null}
agent = Agent(
name="researcher",
max_retries=2, # Retry transient failures
timeout=60, # Prevent hanging
)
```
### Handle Tool Errors
Instruct agents to handle errors gracefully:
```python theme={null}
agent = Agent(
instructions="""
If a tool fails:
1. Try an alternative approach
2. If no alternative, report the error clearly
3. Suggest what information is missing
""",
)
```
***
## Performance Optimization
### Control Iterations
Limit reasoning loops:
```python theme={null}
agent = Agent(
max_iterations=3, # Prevent infinite loops
max_tool_calls=5, # Limit tool usage
timeout=45, # Set timeout
)
```
### Use Appropriate Models
Choose models based on task complexity:
```python theme={null}
# Simple tasks - Fast, cheap model
simple_agent = Agent(
llm=gemini(model="gemini-2.5-flash"), # Fast
)
# Complex reasoning - Better model
complex_agent = Agent(
llm=gemini(model="gemini-2.0-flash-exp"), # More capable
)
```
***
## Testing Agents
### Unit Testing
Test agent logic in isolation:
```python theme={null}
async def test_researcher():
result = await researcher.process_task({
"query": "Test query"
})
assert result["status"] == "success"
assert "summary" in result
```
### Integration Testing
Test with real tools and services:
```python theme={null}
async def test_researcher_integration():
# Use test API keys
result = await researcher.process_task({
"query": "Python async"
})
# Verify tool calls worked
assert len(result["tool_calls"]) > 0
```
***
## Common Pitfalls
### Over-Complex Instructions
Keep instructions focused:
```python theme={null}
# ❌ Bad - Too complex
instructions="""
When processing tasks, first analyze the context, then consider
all possible approaches, evaluate each one, consider edge cases,
think about error handling, plan the execution, execute carefully,
verify results, and finally format the output...
"""
# ✅ Good - Clear and focused
instructions="""
Process the task in 3 steps:
1. Analyze the input
2. Execute the required action
3. Format the result
"""
```
### Conflicting Instructions
Ensure instructions don't conflict:
```python theme={null}
# ❌ Bad - Conflicting
instructions="""
Always be concise. Provide detailed explanations with examples.
"""
# ✅ Good - Consistent
instructions="""
Provide concise summaries with 2-3 key examples when relevant.
"""
```
***
## Next Steps
* [Agent Configuration](/guides/agents/agent-config) - Complete configuration reference
* [Tool Development](/guides/tools/tool-config) - Create custom tools
* [Scaling & Operations](/guides/scaling-and-ops) - Production deployment
* [Examples](/examples/filesystem-agent) - See agent examples
# Local Runtime
Source: https://laddr.agnetlabs.com/guides/local-runtime
Run Laddr locally with in-memory queue and SQLite for testing and development - no Docker required
Run Laddr agents locally using an in-memory queue and SQLite database. No Docker or Redis required — ideal for debugging, development, and performance benchmarking.
***
## Overview
Local runtime mode uses:
* **In-Memory Queue** - `MemoryBus` for single-process communication
* **SQLite Database** - Local file-based storage for traces
* **No External Dependencies** - No Docker, Redis, or PostgreSQL needed
Perfect for quick testing, debugging, and development. For multi-agent workflows with delegation, use Redis or Kafka.
***
## Configuration
Set up your environment file:
```bash theme={null}
# .env
LLM_BACKEND=openai
QUEUE_BACKEND=memory
DB_BACKEND=sqlite
DATABASE_URL=sqlite:///./laddr.db
OPENAI_API_KEY=sk-proj-***
RESEARCHER_MODEL=gpt-4o-mini
COORDINATOR_MODEL=gpt-4o-mini
ANALYZER_MODEL=gpt-4o-mini
WRITER_MODEL=gpt-4o-mini
VALIDATOR_MODEL=gpt-4o-mini
```
No Redis or Docker dependencies are required. SQLite stores traces locally in `laddr.db`.
***
## Running Agents
### Single Agent
Run a single agent locally:
```bash theme={null}
laddr run-local researcher --input '{"query": "What is Laddr?"}'
```
Or using the runner script:
```bash theme={null}
AGENT_NAME=researcher python main.py run '{"query": "Write about AI"}'
```
### Agent with Tools
Run an agent that uses tools:
```bash theme={null}
AGENT_NAME=analyzer python main.py run '{"query": "Calculate 100 + 200 + 300"}'
```
***
## Sequential Workflows
Run multiple agents in sequence:
```python theme={null}
from laddr import AgentRunner, LaddrConfig
import asyncio
import uuid
async def test():
runner = AgentRunner(env_config=LaddrConfig())
job_id = str(uuid.uuid4())
inputs = {'query': 'Calculate 100 + 200 + 300'}
for agent in ['analyzer', 'writer']:
result = await runner.run(inputs, agent_name=agent, job_id=job_id)
if result.get('status') == 'success':
inputs = {'input': result['result']}
asyncio.run(test())
```
***
## Debugging and Traces
### View Traces
Traces are stored in SQLite:
```bash theme={null}
sqlite3 laddr.db "SELECT agent_name, event_type, timestamp FROM traces ORDER BY id DESC LIMIT 10;"
```
### Common Events
Trace events include:
* `task_start` - Task execution started
* `task_complete` - Task execution completed
* `llm_usage` - LLM API call with token usage
* `tool_call` - Tool invocation
* `tool_error` - Tool execution error
* `autonomous_think` - Agent reasoning step
### Query Traces
```bash theme={null}
# View all events for a job
sqlite3 laddr.db "SELECT * FROM traces WHERE job_id = 'your-job-id';"
# Count events by type
sqlite3 laddr.db "SELECT event_type, COUNT(*) FROM traces GROUP BY event_type;"
# View LLM token usage
sqlite3 laddr.db "SELECT agent_name, SUM(tokens_used) FROM traces WHERE event_type = 'llm_usage' GROUP BY agent_name;"
```
***
## Running Workers Locally
### Single Worker
Start a worker process:
```bash theme={null}
python agents/researcher.py
```
### Multiple Workers
Run multiple workers in separate terminals:
```bash theme={null}
# Terminal 1
python agents/coordinator.py
# Terminal 2
python agents/researcher.py
# Terminal 3
python agents/writer.py
```
Delegation (agents handing tasks to other workers) requires a queue backend such as Redis or Kafka to route tasks between processes. MemoryBus only supports single-process communication.
***
## Known Limitations
### MemoryBus Limitations
* ⚠️ **Single Process Only** - `MemoryBus` only works within one process
* ⚠️ **No Inter-Process Delegation** - Can't delegate between separate worker processes
* ⚠️ **No Persistence** - Messages are lost on process restart
### When to Use Memory Backend
✅ **Good for:**
* Single-agent testing
* Debugging agent logic
* Development and prototyping
* Performance benchmarking
❌ **Not suitable for:**
* Multi-agent workflows with delegation
* Production deployments
* Distributed systems
* High availability requirements
***
## Guidelines
### Best Practices
* ✅ Use single-agent mode for debugging
* ✅ Use sequential mode for chained workflows
* ✅ Inspect traces to verify execution
* ✅ Use Redis/Kafka for multi-agent delegation
### Avoid
* 🚫 Don't use delegation without workers
* 🚫 Don't use for production workloads
* 🚫 Don't expect message persistence
***
## Switching to Distributed Mode
When ready for multi-agent workflows:
### Switch to Redis
```bash theme={null}
# .env
QUEUE_BACKEND=redis
REDIS_URL=redis://localhost:6379/0
```
### Switch to Kafka
```bash theme={null}
# .env
QUEUE_BACKEND=kafka
KAFKA_BOOTSTRAP=kafka:9092
```
Then start workers:
```bash theme={null}
# Start Redis
docker run -d -p 6379:6379 redis
# Or start Kafka
docker compose up -d kafka
# Start workers
python agents/researcher.py
python agents/coordinator.py
```
***
## Notes
* 🧠 `MemoryBus` is a singleton that handles agent task routing in the same process
* 🗄️ SQLite logging ensures full trace visibility for debugging
* 🚀 For distributed execution, switch to `QUEUE_BACKEND=redis` or `QUEUE_BACKEND=kafka`
***
## Next Steps
* [Scaling & Operations](/guides/scaling-and-ops) - Production deployment
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
* [Installation](/getting-started/install) - Full setup guide
# Ollama Integration
Source: https://laddr.agnetlabs.com/guides/ollama
Configure and use Ollama for running local LLM models with Laddr agents - Docker setup, configuration, and examples
Run Laddr agents with local LLM models using Ollama. This guide covers configuration, Docker setup, and usage examples.
***
## Configuration
### Environment Variables
Add these to your `.env` file:
```bash theme={null}
# ===================================
# Ollama Configuration
# ===================================
# Ollama server URL
# For local development (Ollama running on host):
OLLAMA_BASE_URL=http://localhost:11434
# For Docker containers (see Docker Setup section):
# OLLAMA_BASE_URL=http://host.docker.internal:11434
# Default Ollama model (used if not specified per-agent)
OLLAMA_MODEL=gemma2:2b
# ===================================
# Backend Selection
# ===================================
# Set Ollama as default LLM backend for all agents
LLM_BACKEND=ollama
# Or configure per-agent:
LLM_BACKEND_RESEARCHER=ollama
LLM_BACKEND_COORDINATOR=openai # Mix with cloud models
# ===================================
# Per-Agent Models
# ===================================
# Use different Ollama models for different agents
LLM_MODEL_RESEARCHER=gemma2:2b # Fast, cheap model
LLM_MODEL_COORDINATOR=llama3.2:latest # Better reasoning
LLM_MODEL_WRITER=mistral:7b # Good at writing
```
### Configuration Priority
Laddr resolves Ollama configuration in this order (highest to lowest priority):
**For base URL:**
1. Per-agent base URL: `LLM_BASE_URL_RESEARCHER=http://custom:11434`
2. Global Ollama URL: `OLLAMA_BASE_URL=http://localhost:11434`
3. Default: `http://localhost:11434`
**For models:**
1. Per-agent model: `LLM_MODEL_RESEARCHER=gemma2:2b`
2. Global Ollama model: `OLLAMA_MODEL=llama3.2:latest`
3. Global LLM model: `LLM_MODEL=gemma2:2b`
4. Default: `gemma2:2b`
***
## Docker Setup
When running Laddr in Docker containers, special network configuration is needed so containers can reach Ollama running on the host machine.
### Problem: Container Network Isolation
By default, `http://localhost:11434` inside a Docker container refers to the **container itself**, not the host machine where Ollama is running.
**Error you might see:**
```text theme={null}
LLM generation failed: Ollama endpoints not reachable at http://localhost:11434
Cannot connect to host localhost:11434
```
### Solution: Use host.docker.internal
Docker provides a special hostname `host.docker.internal` that resolves to the host's IP address.
### Docker Compose Configuration
Update your `docker-compose.yml`:
```yaml theme={null}
services:
api:
build:
context: ..
dockerfile: tester/Dockerfile
ports:
- 8000:8000
environment:
# Point to host's Ollama via special hostname
OLLAMA_BASE_URL: "http://host.docker.internal:11434"
# Map host.docker.internal to host gateway
extra_hosts:
- "host.docker.internal:host-gateway"
# ... other config
researcher_worker:
build:
context: ..
dockerfile: tester/Dockerfile
command: python -m agents.researcher
environment:
AGENT_NAME: researcher
# Worker must also use host.docker.internal
OLLAMA_BASE_URL: "http://host.docker.internal:11434"
extra_hosts:
- "host.docker.internal:host-gateway"
# ... other config
coordinator_worker:
build:
context: ..
dockerfile: tester/Dockerfile
command: python -m agents.coordinator
environment:
AGENT_NAME: coordinator
OLLAMA_BASE_URL: "http://host.docker.internal:11434"
extra_hosts:
- "host.docker.internal:host-gateway"
# ... other config
```
### Key Configuration Points
**1. Environment Variable**
```yaml theme={null}
environment:
OLLAMA_BASE_URL: "http://host.docker.internal:11434"
```
Tells Laddr to connect to Ollama via the special hostname.
**2. Extra Hosts Mapping**
```yaml theme={null}
extra_hosts:
- "host.docker.internal:host-gateway"
```
Maps the hostname to the Docker host's gateway IP.
**3. All Services Need It**
Apply this configuration to **all services** that use Ollama:
* API server
* Worker containers
* Any agent services
### Verification in Docker
Test Ollama connectivity from inside a container:
```bash theme={null}
# Start your containers
docker compose up -d
# Exec into a worker container
docker compose exec researcher_worker bash
# Test connection
curl http://host.docker.internal:11434/api/generate -d '{
"model": "gemma2:2b",
"prompt": "test",
"stream": false
}'
```
Expected: JSON response with generated text.
***
## Usage Examples
### Example 1: Simple Agent with Ollama
```python theme={null}
# my_agent.py
from laddr import Agent
from laddr.llms import ollama
researcher = Agent(
name="researcher",
role="Research Specialist",
goal="Find and analyze information",
backstory="You are an expert researcher.",
# Use Ollama with gemma2:2b model
llm=ollama(model="gemma2:2b"),
tools=[],
instructions="Research the given topic thoroughly."
)
```
### Example 2: Custom Ollama Server
```python theme={null}
from laddr import Agent
from laddr.llms import ollama
# Connect to Ollama on custom port/host
researcher = Agent(
name="researcher",
role="Researcher",
goal="Research information",
backstory="Expert researcher",
llm=ollama(
model="llama3.2:latest",
base_url="http://192.168.1.100:11434" # Remote Ollama server
),
tools=[]
)
```
### Example 3: Mixed Backends
```python theme={null}
from laddr import Agent
from laddr.llms import ollama, gemini
# Coordinator uses cloud API (complex reasoning)
coordinator = Agent(
name="coordinator",
role="Task Coordinator",
goal="Coordinate research tasks",
backstory="Experienced coordinator",
llm=gemini("gemini-2.0-flash"), # Cloud API
is_coordinator=True
)
# Researcher uses local Ollama (cost-effective)
researcher = Agent(
name="researcher",
role="Researcher",
goal="Find information",
backstory="Research specialist",
llm=ollama("gemma2:2b"), # Local model
tools=[web_search]
)
```
### Example 4: Environment-Based Configuration
```python theme={null}
# agents/researcher.py
from laddr import Agent
from laddr.llms import ollama
# Model and URL from environment variables
researcher = Agent(
name="researcher",
role="Researcher",
goal="Research topics",
backstory="Expert researcher",
# Automatically uses:
# - LLM_MODEL_RESEARCHER or OLLAMA_MODEL or default
# - OLLAMA_BASE_URL or default
llm=ollama(),
tools=[]
)
```
***
## Available Models
Popular Ollama models you can use:
* **gemma2:2b** - Fast, lightweight (good for simple tasks)
* **llama3.2:latest** - Balanced performance and quality
* **mistral:7b** - Good for writing and analysis
* **llama3.1:8b** - Strong reasoning capabilities
* **qwen2.5:7b** - Multilingual support
Start with smaller models (2b-7b) for faster responses. Use larger models (13b+) for complex reasoning tasks.
***
## Troubleshooting
### Connection Errors
If you see connection errors:
1. **Verify Ollama is running:**
```bash theme={null}
curl http://localhost:11434/api/tags
```
2. **Check Docker network:**
```bash theme={null}
docker compose exec worker curl http://host.docker.internal:11434/api/tags
```
3. **Verify environment variables:**
```bash theme={null}
docker compose exec worker env | grep OLLAMA
```
### Model Not Found
If a model is not found:
1. **Pull the model:**
```bash theme={null}
ollama pull gemma2:2b
```
2. **List available models:**
```bash theme={null}
ollama list
```
### Performance Issues
If responses are slow:
1. Use smaller models for simple tasks
2. Increase Ollama's context window if needed
3. Check system resources (CPU, RAM)
4. Consider using GPU acceleration
***
## Next Steps
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
* [Local Runtime](/guides/local-runtime) - Local development
* [Scaling & Operations](/guides/scaling-and-ops) - Production deployment
# Scaling & Operations
Source: https://laddr.agnetlabs.com/guides/scaling-and-ops
Scale Laddr agents and deploy to production - horizontal scaling, queue backends, monitoring, and performance tuning
Learn how to scale Laddr agents horizontally and deploy to production environments.
***
## Horizontal Scaling
### Scale Workers
Scale agent workers to handle increased load:
```bash theme={null}
# Scale a single agent type
laddr scale researcher 5
# Scale multiple agents
laddr scale researcher 5
laddr scale coordinator 3
laddr scale writer 2
```
### Docker Compose Scaling
Scale workers using Docker Compose:
```bash theme={null}
# Scale coordinator workers to 3 instances
docker compose up -d --scale coordinator_worker=3
# Scale all workers
docker compose up -d \\
--scale coordinator_worker=3 \\
--scale researcher_worker=3 \\
--scale analyzer_worker=2 \\
--scale writer_worker=2
```
***
## Queue Backends
### Redis (Development)
Fast, lightweight queue backend for development:
```bash theme={null}
# .env
QUEUE_BACKEND=redis
REDIS_URL=redis://localhost:6379/0
```
### Kafka (Production)
Durable, scalable queue backend for production:
```bash theme={null}
# .env
QUEUE_BACKEND=kafka
KAFKA_BOOTSTRAP=kafka:9092
```
Kafka provides better message persistence and horizontal scaling capabilities for production workloads.
### Memory (Testing)
In-memory queue for local testing:
```bash theme={null}
# .env
QUEUE_BACKEND=memory
```
Memory backend only works within a single process. Use Redis or Kafka for multi-worker deployments.
***
## Database Configuration
### PostgreSQL (Production)
Use PostgreSQL for production deployments:
```bash theme={null}
# .env
DB_BACKEND=postgresql
DATABASE_URL=postgresql://user:password@localhost:5432/laddr
```
### SQLite (Development)
SQLite for local development:
```bash theme={null}
# .env
DB_BACKEND=sqlite
DATABASE_URL=sqlite:///./laddr.db
```
***
## Monitoring
### Dashboard
Access the dashboard for real-time monitoring:
```bash theme={null}
# Start dashboard
laddr run dev -d
# Access at http://localhost:5173
```
### Metrics
Monitor key metrics:
* **Queue Depth** - Number of pending tasks
* **Worker Utilization** - Active workers vs idle
* **Throughput** - Tasks processed per second
* **Error Rate** - Failed tasks percentage
* **Latency** - Average task completion time
### Logs
View and follow logs:
```bash theme={null}
# Follow logs for an agent
laddr logs researcher --follow
# Show last 100 lines
laddr logs researcher --tail 100
# View all service logs
docker compose logs -f
```
***
## Production Deployment
### Environment Variables
Configure production environment:
```bash theme={null}
# .env.production
# Queue
QUEUE_BACKEND=kafka
KAFKA_BOOTSTRAP=kafka-cluster:9092
# Database
DB_BACKEND=postgresql
DATABASE_URL=postgresql://user:pass@db-host:5432/laddr
# Storage
STORAGE_BACKEND=s3
AWS_ACCESS_KEY_ID=...
AWS_SECRET_ACCESS_KEY=...
AWS_REGION=us-east-1
# LLM
LLM_PROVIDER=openai
OPENAI_API_KEY=...
```
### Health Checks
Implement health checks:
```bash theme={null}
# Check system health
laddr check
# API health endpoint
curl http://localhost:8000/api/health
```
### Resource Limits
Set appropriate resource limits:
```yaml theme={null}
# docker-compose.yml
services:
researcher_worker:
deploy:
resources:
limits:
cpus: '2'
memory: 2G
reservations:
cpus: '1'
memory: 1G
```
***
## Load Balancing
### Worker Distribution
Kafka automatically distributes tasks across workers:
```mermaid theme={null}
graph LR
Queue[Task Queue] --> W1[Worker 1]
Queue --> W2[Worker 2]
Queue --> W3[Worker 3]
Queue --> W4[Worker 4]
```
Each worker in a consumer group processes a subset of tasks.
### Partition Strategy
Configure Kafka partitions for better parallelism:
```bash theme={null}
# More partitions = more parallelism
# Create topic with 10 partitions
kafka-topics --create \\
--bootstrap-server localhost:9092 \\
--topic laddr.tasks.researcher \\
--partitions 10 \\
--replication-factor 1
```
***
## Performance Tuning
### Worker Configuration
Optimize worker settings:
```bash theme={null}
# .env
MAX_CONCURRENT_TASKS=5 # Tasks per worker
WORKER_PREFETCH=10 # Prefetch count
WORKER_TIMEOUT=300 # Task timeout
```
### Database Connection Pooling
Configure connection pooling:
```bash theme={null}
# .env
DB_POOL_SIZE=20
DB_MAX_OVERFLOW=10
```
***
## Troubleshooting
### High Queue Depth
If queue depth is growing:
1. Scale up workers: `laddr scale researcher 10`
2. Check worker logs for errors
3. Verify database/storage connectivity
4. Check for slow tools or LLM calls
### Worker Failures
If workers are failing:
1. Check logs: `laddr logs researcher --tail 100`
2. Verify API keys and credentials
3. Check resource limits (CPU/memory)
4. Review error messages in dashboard
### Performance Issues
If performance is slow:
1. Monitor dashboard metrics
2. Check database query performance
3. Review LLM response times
4. Optimize tool implementations
5. Consider caching strategies
***
## Next Steps
* [Local Runtime](/guides/local-runtime) - Local development
* [Storage & Artifacts](/guides/storage-and-artifacts) - Configure storage
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
# Storage & Artifacts
Source: https://laddr.agnetlabs.com/guides/storage-and-artifacts
Configure storage backends and manage artifacts in Laddr - MinIO, S3, local storage, and best practices
Laddr supports multiple storage backends for storing large data artifacts, files, and agent outputs.
***
## Storage Backends
### MinIO (S3-Compatible)
MinIO is an S3-compatible object storage service, ideal for development and production.
```bash theme={null}
# .env
STORAGE_BACKEND=minio
MINIO_ENDPOINT=localhost:9000
MINIO_ACCESS_KEY=minioadmin
MINIO_SECRET_KEY=minioadmin
MINIO_SECURE=false # Set to true for HTTPS
```
### AWS S3
Use AWS S3 for production deployments:
```bash theme={null}
# .env
STORAGE_BACKEND=s3
AWS_ACCESS_KEY_ID=your_access_key
AWS_SECRET_ACCESS_KEY=your_secret_key
AWS_REGION=us-east-1
S3_BUCKET=laddr-artifacts
```
### Local File System
For development and testing:
```bash theme={null}
# .env
STORAGE_BACKEND=local
STORAGE_PATH=./storage
```
***
## Artifact Storage
### Storing Artifacts
Use the artifact storage system tool to store large data:
```python theme={null}
from laddr import Agent
# Artifacts are automatically stored when using system tools
result = await agent.delegate_task(
agent_name="processor",
task_description="Process large dataset",
task="process",
task_data={"data": large_dataset} # Automatically stored if large
)
```
### Custom Artifact Storage
Override artifact storage for custom behavior:
```python theme={null}
from laddr import override_system_tool, ArtifactStorageTool
@override_system_tool("system_store_artifact")
async def custom_storage(
data: dict,
artifact_type: str,
metadata: dict = None,
bucket: str = None,
_artifact_storage=None,
**kwargs
):
"""Custom artifact storage with compression."""
storage_tool = ArtifactStorageTool(
storage_backend=_artifact_storage,
default_bucket=bucket or "artifacts"
)
# Add custom logic (compression, encryption, etc.)
return await storage_tool.store_artifact(
data=data,
artifact_type=artifact_type,
metadata=metadata,
bucket=bucket
)
```
***
## Configuration
### Bucket Configuration
Configure default buckets:
```bash theme={null}
# .env
STORAGE_DEFAULT_BUCKET=artifacts
STORAGE_ARTIFACTS_BUCKET=laddr-artifacts
STORAGE_LOGS_BUCKET=laddr-logs
```
### Storage Limits
Set storage limits and retention:
```bash theme={null}
# .env
STORAGE_MAX_SIZE=1073741824 # 1GB in bytes
STORAGE_RETENTION_DAYS=30
```
***
## Best Practices
### 1. Use Appropriate Backends
* **Development**: Local file system or MinIO
* **Production**: AWS S3 or MinIO cluster
* **Testing**: In-memory or local file system
### 2. Organize by Bucket
Use different buckets for different artifact types:
```python theme={null}
# Store different types in different buckets
await store_artifact(data, artifact_type="result", bucket="results")
await store_artifact(data, artifact_type="log", bucket="logs")
await store_artifact(data, artifact_type="cache", bucket="cache")
```
### 3. Set Retention Policies
Configure automatic cleanup:
```bash theme={null}
# .env
STORAGE_RETENTION_DAYS=30 # Delete artifacts older than 30 days
STORAGE_CLEANUP_INTERVAL=3600 # Run cleanup every hour
```
***
## Next Steps
* [System Tools](/guides/system-tools/custom-system-tools) - Customize artifact storage
* [Scaling & Operations](/guides/scaling-and-ops) - Production deployment
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
# Custom System Tools
Source: https://laddr.agnetlabs.com/guides/system-tools/custom-system-tools
Complete guide to creating custom system tool overrides using Laddr's base tool classes - patterns, examples, and best practices
This guide explains how to create custom system tools by importing and extending Laddr's base tool classes.
***
## Overview
Laddr provides three base system tool classes that you can import and use to build custom overrides:
* **`TaskDelegationTool`** - Single-task delegation to other agents
* **`ParallelDelegationTool`** - Parallel multi-task delegation (fan-out pattern)
* **`ArtifactStorageTool`** - Storage and retrieval of large data artifacts
These classes are fully composable, meaning you can use them as building blocks within your custom tool implementations.
***
## Importing Base Tools
You can import the base tools from three different locations:
### Top-Level Import (Recommended)
```python theme={null}
from laddr import TaskDelegationTool, ParallelDelegationTool, ArtifactStorageTool
```
### Core Module Import
```python theme={null}
from laddr.core import TaskDelegationTool, ParallelDelegationTool, ArtifactStorageTool
```
### Direct Module Import
```python theme={null}
from laddr.core.system_tools import TaskDelegationTool, ParallelDelegationTool, ArtifactStorageTool
```
***
## Importing the Override Decorator
To register your custom tools, you'll need the override decorator:
```python theme={null}
from laddr import override_system_tool
# or
from laddr.core import override_system_tool
```
***
## Understanding System Tool Architecture
### Runtime Injection
When your custom tool override is called, the Laddr runtime automatically injects three special keyword arguments:
* `_message_bus` - The message queue backend (Redis/Kafka)
* `_artifact_storage` - The storage backend (MinIO/S3)
* `_agent` - The current agent instance (for job\_id propagation)
These are injected by the runtime - you don't pass them manually.
### Base Tool Initialization
All base tool classes accept these dependencies in their constructors:
```python theme={null}
# TaskDelegationTool and ParallelDelegationTool
tool = TaskDelegationTool(
message_bus=_message_bus,
artifact_storage=_artifact_storage,
agent=_agent
)
# ArtifactStorageTool
tool = ArtifactStorageTool(
storage_backend=_artifact_storage,
default_bucket="my-bucket"
)
```
***
## Common Patterns
### Pattern 1: Enhance with Pre/Post Processing
Add logging, metrics, or validation while reusing base functionality:
```python theme={null}
from laddr import override_system_tool, TaskDelegationTool
import logging
logger = logging.getLogger(__name__)
@override_system_tool("system_delegate_parallel")
async def custom_parallel_delegation(
agent_name: str,
tasks: list[dict],
timeout_seconds: int = 300,
_message_bus=None,
_artifact_storage=None,
_agent=None
):
"""Enhanced parallel delegation with logging and metrics."""
# Pre-processing
logger.info(f"Starting parallel delegation of {len(tasks)} tasks to {agent_name}")
# Reuse base TaskDelegationTool for actual delegation
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
results = []
for task in tasks:
logger.debug(f"Delegating task: {task.get('task_description', 'N/A')}")
result = await delegation_tool.delegate_task(
agent_name=agent_name,
task_description=task.get("task_description", ""),
task=task.get("task"),
task_data=task.get("task_data"),
timeout_seconds=timeout_seconds
)
results.append(result)
# Post-processing
logger.info(f"Completed {len(results)} delegations")
return {"results": results, "total": len(results)}
```
### Pattern 2: Add Rate Limiting
Control delegation rate to prevent overwhelming downstream agents:
```python theme={null}
from laddr import override_system_tool, TaskDelegationTool
import asyncio
import time
class RateLimiter:
def __init__(self, max_per_second: int):
self.max_per_second = max_per_second
self.last_call = 0
async def wait(self):
now = time.time()
elapsed = now - self.last_call
if elapsed < (1.0 / self.max_per_second):
await asyncio.sleep((1.0 / self.max_per_second) - elapsed)
self.last_call = time.time()
rate_limiter = RateLimiter(max_per_second=5)
@override_system_tool("system_delegate_task")
async def rate_limited_delegation(
agent_name: str,
task_description: str,
task: str,
task_data: dict = None,
timeout_seconds: int = 300,
_message_bus=None,
_artifact_storage=None,
_agent=None
):
"""Rate-limited task delegation."""
# Wait for rate limiter
await rate_limiter.wait()
# Use base tool for actual delegation
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
return await delegation_tool.delegate_task(
agent_name=agent_name,
task_description=task_description,
task=task,
task_data=task_data,
timeout_seconds=timeout_seconds
)
```
### Pattern 3: Add Retry Logic
Implement automatic retries for failed delegations:
```python theme={null}
from laddr import override_system_tool, TaskDelegationTool
import asyncio
import logging
logger = logging.getLogger(__name__)
@override_system_tool("system_delegate_task")
async def retry_delegation(
agent_name: str,
task_description: str,
task: str,
task_data: dict = None,
timeout_seconds: int = 300,
max_retries: int = 3,
_message_bus=None,
_artifact_storage=None,
_agent=None
):
"""Task delegation with automatic retries."""
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
for attempt in range(max_retries):
try:
result = await delegation_tool.delegate_task(
agent_name=agent_name,
task_description=task_description,
task=task,
task_data=task_data,
timeout_seconds=timeout_seconds
)
return result
except Exception as e:
logger.warning(f"Delegation attempt {attempt + 1} failed: {e}")
if attempt == max_retries - 1:
logger.error(f"All {max_retries} attempts failed")
raise
# Exponential backoff
await asyncio.sleep(2 ** attempt)
```
### Pattern 4: Add Circuit Breaker
Prevent cascading failures by stopping delegation to failing agents:
```python theme={null}
from laddr import override_system_tool, TaskDelegationTool
from collections import defaultdict
import time
class CircuitBreaker:
def __init__(self, failure_threshold: int = 5, timeout: int = 60):
self.failure_threshold = failure_threshold
self.timeout = timeout
self.failures = defaultdict(int)
self.last_failure_time = defaultdict(float)
def is_open(self, agent_name: str) -> bool:
"""Check if circuit breaker is open for this agent."""
if self.failures[agent_name] < self.failure_threshold:
return False
# Check if timeout has elapsed
if time.time() - self.last_failure_time[agent_name] > self.timeout:
# Reset circuit breaker
self.failures[agent_name] = 0
return False
return True
def record_failure(self, agent_name: str):
"""Record a failure for this agent."""
self.failures[agent_name] += 1
self.last_failure_time[agent_name] = time.time()
def record_success(self, agent_name: str):
"""Record a success for this agent."""
self.failures[agent_name] = 0
circuit_breaker = CircuitBreaker(failure_threshold=5, timeout=60)
@override_system_tool("system_delegate_task")
async def circuit_breaker_delegation(
agent_name: str,
task_description: str,
task: str,
task_data: dict = None,
timeout_seconds: int = 300,
_message_bus=None,
_artifact_storage=None,
_agent=None
):
"""Task delegation with circuit breaker pattern."""
# Check circuit breaker
if circuit_breaker.is_open(agent_name):
raise RuntimeError(f"Circuit breaker open for agent '{agent_name}'")
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
try:
result = await delegation_tool.delegate_task(
agent_name=agent_name,
task_description=task_description,
task=task,
task_data=task_data,
timeout_seconds=timeout_seconds
)
circuit_breaker.record_success(agent_name)
return result
except Exception as e:
circuit_breaker.record_failure(agent_name)
raise
```
### Pattern 5: Custom Artifact Storage with Compression
Add automatic compression for large artifacts:
```python theme={null}
from laddr import override_system_tool, ArtifactStorageTool
import gzip
import json
@override_system_tool("system_store_artifact")
async def compressed_storage(
data: dict,
artifact_type: str,
metadata: dict = None,
bucket: str = None,
compress: bool = True,
_artifact_storage=None,
**kwargs
):
"""Store artifacts with optional compression."""
storage_tool = ArtifactStorageTool(
storage_backend=_artifact_storage,
default_bucket=bucket or "artifacts"
)
# Compress if enabled and data is large
if compress and len(json.dumps(data)) > 1024: # Compress if > 1KB
compressed_data = gzip.compress(json.dumps(data).encode())
# Update metadata to indicate compression
metadata = metadata or {}
metadata["compressed"] = True
metadata["original_size"] = len(json.dumps(data))
metadata["compressed_size"] = len(compressed_data)
# Store compressed data
return await storage_tool.store_artifact(
data={"compressed": compressed_data.hex()},
artifact_type=artifact_type,
metadata=metadata,
bucket=bucket
)
else:
# Store uncompressed
return await storage_tool.store_artifact(
data=data,
artifact_type=artifact_type,
metadata=metadata,
bucket=bucket
)
```
***
## System Tool Reference
### TaskDelegationTool
Handles single-task delegation to other agents.
```python theme={null}
class TaskDelegationTool:
def __init__(
self,
message_bus: MessageBus,
artifact_storage: Storage,
agent: Agent
):
"""
Initialize task delegation tool.
Args:
message_bus: Message queue backend
artifact_storage: Storage backend for large payloads
agent: Current agent instance
"""
pass
async def delegate_task(
self,
agent_name: str,
task_description: str,
task: str,
task_data: dict = None,
timeout_seconds: int = 300
) -> dict:
"""
Delegate a task to another agent.
Args:
agent_name: Name of target agent
task_description: Human-readable task description
task: Task type/action identifier
task_data: Additional task parameters
timeout_seconds: Maximum wait time for result
Returns:
Task result from target agent
"""
pass
```
### ParallelDelegationTool
Handles parallel multi-task delegation (fan-out pattern).
```python theme={null}
class ParallelDelegationTool:
def __init__(
self,
message_bus: MessageBus,
artifact_storage: Storage,
agent: Agent
):
"""
Initialize parallel delegation tool.
Args:
message_bus: Message queue backend
artifact_storage: Storage backend
agent: Current agent instance
"""
pass
async def delegate_parallel(
self,
agent_name: str,
tasks: list[dict],
timeout_seconds: int = 300
) -> dict:
"""
Delegate multiple tasks in parallel to an agent.
Args:
agent_name: Name of target agent
tasks: List of task dictionaries with 'task' and 'task_data' keys
timeout_seconds: Maximum wait time per task
Returns:
Dictionary with 'results' list and 'total' count
"""
pass
```
### ArtifactStorageTool
Handles storage and retrieval of large data artifacts.
```python theme={null}
class ArtifactStorageTool:
def __init__(
self,
storage_backend: Storage,
default_bucket: str = "artifacts"
):
"""
Initialize artifact storage tool.
Args:
storage_backend: Storage backend (MinIO/S3)
default_bucket: Default bucket name
"""
pass
async def store_artifact(
self,
data: dict,
artifact_type: str,
metadata: dict = None,
bucket: str = None
) -> dict:
"""
Store a data artifact.
Args:
data: Data to store
artifact_type: Type/category of artifact
metadata: Additional metadata
bucket: Storage bucket (optional)
Returns:
Dictionary with artifact_id and storage location
"""
pass
async def retrieve_artifact(
self,
artifact_id: str,
bucket: str = None
) -> dict:
"""
Retrieve a stored artifact.
Args:
artifact_id: Unique artifact identifier
bucket: Storage bucket (optional)
Returns:
Original artifact data
"""
pass
```
***
## Available System Tools to Override
You can override any of the following system tools:
* `system_delegate_task` - Single task delegation
* `system_delegate_parallel` - Parallel task delegation
* `system_store_artifact` - Store data artifact
* `system_retrieve_artifact` - Retrieve data artifact
***
## Registering Your Overrides
### In Agent Files
Define your overrides in your agent Python file:
```python theme={null}
from laddr import Agent, override_system_tool, TaskDelegationTool
@override_system_tool("system_delegate_task")
async def my_custom_delegate(
agent_name: str,
task_description: str,
task: str,
task_data: dict = None,
timeout_seconds: int = 300,
_message_bus=None,
_artifact_storage=None,
_agent=None
):
"""My custom delegation logic."""
# Your custom logic here
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
return await delegation_tool.delegate_task(...)
class MyAgent(Agent):
"""Your agent implementation."""
pass
```
### In Separate Override Modules
Create a dedicated module for your overrides:
```python theme={null}
# overrides/delegation.py
from laddr import override_system_tool, TaskDelegationTool
@override_system_tool("system_delegate_task")
async def custom_delegate(...):
"""Custom delegation."""
pass
@override_system_tool("system_delegate_parallel")
async def custom_parallel_delegate(...):
"""Custom parallel delegation."""
pass
```
Then import in your agent:
```python theme={null}
# my_agent.py
from laddr import Agent
import overrides.delegation # Registers overrides on import
class MyAgent(Agent):
pass
```
***
## Testing Your Overrides
### Verify Override Registration
```python theme={null}
from laddr.core.system_tools import get_tool_override, list_tool_overrides
# Check if your override is registered
override_func = get_tool_override("system_delegate_task")
print(f"Override registered: {override_func is not None}")
# List all registered overrides
all_overrides = list_tool_overrides()
print(f"All overrides: {all_overrides}")
```
### Clear Overrides (for testing)
```python theme={null}
from laddr.core.system_tools import clear_tool_overrides
# Clear all overrides (useful in test cleanup)
clear_tool_overrides()
```
***
## Best Practices
### 1. Always Accept Runtime-Injected Parameters
Your override functions must accept these parameters:
```python theme={null}
async def my_override(
# ... your parameters ...
_message_bus=None, # Always include
_artifact_storage=None, # Always include
_agent=None # Always include
):
pass
```
### 2. Reuse Base Tools When Possible
Don't reimplement delegation logic - use the base tools:
```python theme={null}
# ❌ Bad - reimplementing delegation
async def my_override(...):
# Manually publish to message bus, handle responses, etc.
pass
# ✅ Good - reusing base tool
async def my_override(..., _message_bus=None, _artifact_storage=None, _agent=None):
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
return await delegation_tool.delegate_task(...)
```
### 3. Preserve Function Signatures
Maintain compatibility with the original system tool signatures.
### 4. Handle Errors Gracefully
```python theme={null}
import logging
logger = logging.getLogger(__name__)
@override_system_tool("system_delegate_task")
async def my_override(...):
try:
# Your logic
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
result = await delegation_tool.delegate_task(...)
return result
except Exception as e:
logger.error(f"Delegation failed: {e}", exc_info=True)
# Optionally re-raise or return error result
return {"error": str(e), "success": False}
```
### 5. Add Comprehensive Logging
```python theme={null}
import logging
logger = logging.getLogger(__name__)
@override_system_tool("system_delegate_task")
async def my_override(agent_name: str, task_description: str, ...):
logger.info(f"Custom delegation to {agent_name}: {task_description}")
try:
result = await ...
logger.info(f"Delegation succeeded: {result}")
return result
except Exception as e:
logger.error(f"Delegation failed: {e}")
raise
```
***
## Troubleshooting
### Override Not Being Called
1. **Check registration**:
```python theme={null}
from laddr.core.system_tools import get_tool_override
override = get_tool_override("system_delegate_task")
print(f"Registered: {override is not None}")
```
2. **Verify import**: Make sure your override module is imported before the agent runs
3. **Check function signature**: Ensure your override has the required parameters
### Runtime Injection Not Working
Make sure you're accepting the injected parameters:
```python theme={null}
# ✅ Correct
async def my_override(..., _message_bus=None, _artifact_storage=None, _agent=None):
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
# ❌ Wrong - missing injected parameters
async def my_override(...):
delegation_tool = TaskDelegationTool(???, ???, ???) # No dependencies!
```
### Base Tool Not Working
Ensure you're passing the injected dependencies correctly:
```python theme={null}
# ✅ Correct
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
# ❌ Wrong - wrong parameter names
delegation_tool = TaskDelegationTool(message_bus, storage, agent) # Undefined!
# ❌ Wrong - wrong order
delegation_tool = TaskDelegationTool(_agent, _message_bus, _artifact_storage)
```
***
## Next Steps
* [Quickstart Guide](/guides/system-tools/custom-system-tools-quickstart) - Get started in 5 minutes
* [Migration Guide](/migration/migration-system-tools) - Upgrade existing overrides
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
# Custom System Tools Quickstart
Source: https://laddr.agnetlabs.com/guides/system-tools/custom-system-tools-quickstart
Get started with custom system tools in 5 minutes - quick examples and common use cases
Learn how to override Laddr's system tools to add custom behavior like logging, metrics, and rate limiting.
***
## What are System Tools?
System tools are built-in capabilities that agents use for:
* **Task Delegation** - Delegate tasks to other agents
* **Parallel Delegation** - Run multiple tasks in parallel
* **Artifact Storage** - Store and retrieve large data
You can override these to add custom behavior while reusing core functionality.
***
## Quick Example: Add Logging
Override task delegation to add logging:
```python theme={null}
from laddr import override_system_tool, TaskDelegationTool
import logging
logger = logging.getLogger(__name__)
@override_system_tool("system_delegate_task")
async def logged_delegation(
agent_name: str,
task_description: str,
task: str,
task_data: dict = None,
timeout_seconds: int = 300,
_message_bus=None,
_artifact_storage=None,
_agent=None
):
"""Delegation with logging."""
logger.info(f"Delegating to {agent_name}: {task_description}")
# Reuse base tool for actual delegation
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
result = await delegation_tool.delegate_task(
agent_name=agent_name,
task_description=task_description,
task=task,
task_data=task_data,
timeout_seconds=timeout_seconds
)
logger.info(f"Delegation complete: {result}")
return result
```
That's it! Your custom delegation is now active.
***
## Common Use Cases
### 1. Add Metrics
Track delegation metrics:
```python theme={null}
from laddr import override_system_tool, TaskDelegationTool
import time
metrics = {"count": 0, "total_time": 0.0}
@override_system_tool("system_delegate_task")
async def metered_delegation(
agent_name: str,
task_description: str,
task: str,
task_data: dict = None,
timeout_seconds: int = 300,
_message_bus=None,
_artifact_storage=None,
_agent=None
):
"""Delegation with metrics."""
start_time = time.time()
metrics["count"] += 1
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
result = await delegation_tool.delegate_task(
agent_name=agent_name,
task_description=task_description,
task=task,
task_data=task_data,
timeout_seconds=timeout_seconds
)
duration = time.time() - start_time
metrics["total_time"] += duration
return result
```
### 2. Add Rate Limiting
Control delegation rate:
```python theme={null}
from laddr import override_system_tool, TaskDelegationTool
import asyncio
import time
class RateLimiter:
def __init__(self, max_per_second: int):
self.max_per_second = max_per_second
self.last_call = 0
async def wait(self):
now = time.time()
elapsed = now - self.last_call
if elapsed < (1.0 / self.max_per_second):
await asyncio.sleep((1.0 / self.max_per_second) - elapsed)
self.last_call = time.time()
rate_limiter = RateLimiter(max_per_second=5)
@override_system_tool("system_delegate_task")
async def rate_limited_delegation(
agent_name: str,
task_description: str,
task: str,
task_data: dict = None,
timeout_seconds: int = 300,
_message_bus=None,
_artifact_storage=None,
_agent=None
):
"""Rate-limited delegation."""
await rate_limiter.wait()
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
return await delegation_tool.delegate_task(
agent_name=agent_name,
task_description=task_description,
task=task,
task_data=task_data,
timeout_seconds=timeout_seconds
)
```
### 3. Add Retries
Automatic retry on failure:
```python theme={null}
from laddr import override_system_tool, TaskDelegationTool
import asyncio
import logging
logger = logging.getLogger(__name__)
@override_system_tool("system_delegate_task")
async def retry_delegation(
agent_name: str,
task_description: str,
task: str,
task_data: dict = None,
timeout_seconds: int = 300,
max_retries: int = 3,
_message_bus=None,
_artifact_storage=None,
_agent=None
):
"""Delegation with automatic retries."""
delegation_tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
for attempt in range(max_retries):
try:
result = await delegation_tool.delegate_task(
agent_name=agent_name,
task_description=task_description,
task=task,
task_data=task_data,
timeout_seconds=timeout_seconds
)
return result
except Exception as e:
logger.warning(f"Attempt {attempt + 1} failed: {e}")
if attempt == max_retries - 1:
raise
# Exponential backoff
await asyncio.sleep(2 ** attempt)
```
***
## Available System Tools
You can override these system tools:
| Tool | Purpose | Base Class |
| -------------------------- | ------------------------------ | ------------------------ |
| `system_delegate_task` | Single task delegation | `TaskDelegationTool` |
| `system_delegate_parallel` | Parallel multi-task delegation | `ParallelDelegationTool` |
| `system_store_artifact` | Store data artifact | `ArtifactStorageTool` |
| `system_retrieve_artifact` | Retrieve data artifact | `ArtifactStorageTool` |
***
## Import Patterns
Import base tools in three ways:
```python theme={null}
# Top-level (recommended)
from laddr import TaskDelegationTool, ParallelDelegationTool, ArtifactStorageTool
# Core module
from laddr.core import TaskDelegationTool, ParallelDelegationTool, ArtifactStorageTool
# Direct module
from laddr.core.system_tools import TaskDelegationTool, ParallelDelegationTool, ArtifactStorageTool
```
***
## Common Mistakes
### ❌ Missing Runtime Parameters
```python theme={null}
# ❌ Wrong - Missing injected parameters
@override_system_tool("system_delegate_task")
async def bad_override(agent_name: str, task: str):
# Can't use base tools without dependencies!
pass
# ✅ Correct - Include all parameters
@override_system_tool("system_delegate_task")
async def good_override(
agent_name: str,
task: str,
_message_bus=None,
_artifact_storage=None,
_agent=None
):
tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
return await tool.delegate_task(...)
```
### ❌ Reimplementing Core Logic
```python theme={null}
# ❌ Bad - Reimplementing everything
@override_system_tool("system_delegate_task")
async def bad_override(...):
# 100+ lines of message bus logic
# Easy to introduce bugs
pass
# ✅ Good - Reuse base tool
@override_system_tool("system_delegate_task")
async def good_override(..., _message_bus=None, _artifact_storage=None, _agent=None):
tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
return await tool.delegate_task(...)
```
***
## Next Steps
* [Complete Guide](/guides/system-tools/custom-system-tools) - Full documentation
* [Migration Guide](/migration/migration-system-tools) - Upgrade existing overrides
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
# MCP Integration
Source: https://laddr.agnetlabs.com/guides/tools/mcp-integration
Connect Laddr agents to Model Context Protocol (MCP) servers - transport methods, lifecycle, and examples
The [Model Context Protocol (MCP)](https://modelcontextprotocol.io) enables agents to interact with external systems through a standardized interface. Connect your agents to any MCP server using Laddr's MCP integration.
***
## Quick Start
Connect an agent to an MCP server:
```python theme={null}
from laddr import Agent, WorkerRunner
from laddr.llms import gemini
from laddr.core.mcp_tools import MCPToolProvider
# Create MCP tool provider
mcp_provider = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-filesystem /path/to/directory",
transport="stdio"
)
# Create the Agent with MCP tools
agent = Agent(
name="filesystem_agent",
role="Filesystem Assistant",
goal="Help users explore files and directories",
backstory="A helpful assistant that can navigate and analyze files",
llm=gemini(model="gemini-2.5-flash"),
tools=[mcp_provider], # MCP tools are automatically discovered and registered
)
# Run as worker
async def main():
runner = WorkerRunner(agent=agent)
await runner.start()
if __name__ == "__main__":
import asyncio
asyncio.run(main())
```
***
## Basic Usage
### Single MCP Server
The simplest way to use MCP is with a single server:
```python theme={null}
from laddr import Agent
from laddr.core.mcp_tools import MCPToolProvider
# Stdio transport (local command-based server)
mcp = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-filesystem /path",
transport="stdio"
)
agent = Agent(
name="researcher",
role="Research Agent",
goal="Research topics using MCP tools",
tools=[mcp], # MCP tools automatically available
# ... other agent config
)
```
### Multiple MCP Servers
Use `MultiMCPToolProvider` to connect to multiple servers:
```python theme={null}
from laddr import Agent
from laddr.core.mcp_tools import MCPToolProvider, MultiMCPToolProvider
# Create multiple providers
filesystem_mcp = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-filesystem /path",
transport="stdio",
server_name="filesystem"
)
git_mcp = MCPToolProvider(
command="uvx mcp-server-git",
transport="stdio",
server_name="git"
)
# Combine into multi-provider
multi_mcp = MultiMCPToolProvider([filesystem_mcp, git_mcp])
agent = Agent(
name="developer",
role="Developer Assistant",
goal="Help with filesystem and git operations",
tools=[multi_mcp],
# ... other agent config
)
```
***
## Transport Methods
Laddr supports three MCP transport methods:
### 1. Stdio Transport
For local command-based MCP servers:
```python theme={null}
mcp = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-filesystem /path/to/dir",
transport="stdio"
)
```
**Use when:**
* Running local MCP servers
* Server is launched via command line
* Examples: filesystem, git, database servers
### 2. HTTP Transport (Streamable HTTP)
For remote HTTP-based MCP servers:
```python theme={null}
mcp = MCPToolProvider(
url="https://docs.agno.com/mcp",
transport="streamable-http", # or "http"
api_key="your-api-key" # Optional
)
```
**Use when:**
* Connecting to remote MCP servers
* Server exposes HTTP endpoint
* Examples: hosted MCP services, cloud-based tools
### 3. SSE Transport (Server-Sent Events)
For real-time streaming MCP servers:
```python theme={null}
mcp = MCPToolProvider(
url="https://mcp-server.example.com",
transport="sse",
api_key="your-api-key" # Optional
)
```
**Use when:**
* Server uses Server-Sent Events
* Real-time event streaming needed
* Examples: live data feeds, streaming APIs
***
## Connection Lifecycle
### Automatic Connection Management
When you pass `MCPToolProvider` to an Agent, connections are managed automatically:
```python theme={null}
mcp = MCPToolProvider(command="npx -y @modelcontextprotocol/server-filesystem /path")
agent = Agent(
name="agent",
tools=[mcp],
# ... config
)
# Connection established automatically when agent first uses tools
```
### Manual Connection Management
For more control, use `connect()` and `disconnect()`:
```python theme={null}
mcp = MCPToolProvider(command="npx -y @modelcontextprotocol/server-filesystem /path")
# Connect explicitly
await mcp.connect()
try:
agent = Agent(name="agent", tools=[mcp], ...)
# Use agent...
finally:
# Disconnect when done
await mcp.disconnect()
```
### Async Context Manager
You can also use MCP providers as async context managers:
```python theme={null}
async with MCPToolProvider(command="npx -y @modelcontextprotocol/server-filesystem /path") as mcp:
agent = Agent(name="agent", tools=[mcp], ...)
# Connection automatically closed when exiting context
```
### Auto Refresh
Enable `auto_refresh` to reconnect and refresh tools on each agent run:
```python theme={null}
mcp = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-filesystem /path",
auto_refresh=True # Reconnect and refresh tools on each run
)
```
**Use when:**
* MCP server tools change frequently
* Server may restart between runs
* Tool schema updates are common
***
## Tool Name Collision Handling
When multiple MCP servers provide tools with the same name, Laddr automatically prefixes tool names with the server identifier:
```python theme={null}
# Server 1 provides: read_file
# Server 2 provides: read_file
# Result:
# - filesystem_read_file (from server_name="filesystem")
# - git_read_file (from server_name="git")
```
If `server_name` is not provided, the transport type is used as prefix (e.g., `stdio_read_file`).
***
## Example: Filesystem Agent
Complete example of a filesystem exploration agent:
```python theme={null}
import asyncio
from laddr import Agent, WorkerRunner
from laddr.llms import gemini
from laddr.core.mcp_tools import MCPToolProvider
async def main():
# Initialize MCP provider for filesystem
mcp = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-filesystem /path/to/project",
transport="stdio",
server_name="filesystem"
)
# Create agent
agent = Agent(
name="filesystem_assistant",
role="Filesystem Assistant",
goal="Help users explore and analyze files and directories",
backstory="""You are a helpful filesystem assistant. You can:
- Navigate directories
- Read and analyze files
- Search for content
- Provide file information""",
llm=gemini(model="gemini-2.5-flash", temperature=0.3),
tools=[mcp],
instructions="""
When helping users:
1. Use filesystem tools to explore directories
2. Read relevant files to answer questions
3. Provide clear, organized information
4. Be concise and focus on what the user asked
"""
)
# Run as worker
runner = WorkerRunner(agent=agent)
print("Starting filesystem assistant...")
await runner.start()
if __name__ == "__main__":
asyncio.run(main())
```
***
## Example: Multiple MCP Servers
Combining filesystem and git MCP servers:
```python theme={null}
from laddr import Agent
from laddr.core.mcp_tools import MCPToolProvider, MultiMCPToolProvider
# Filesystem MCP
fs_mcp = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-filesystem .",
transport="stdio",
server_name="filesystem"
)
# Git MCP
git_mcp = MCPToolProvider(
command="uvx mcp-server-git",
transport="stdio",
server_name="git"
)
# Combine
multi_mcp = MultiMCPToolProvider([fs_mcp, git_mcp])
agent = Agent(
name="developer_assistant",
role="Developer Assistant",
goal="Help with codebase exploration and git operations",
tools=[multi_mcp],
# ... other config
)
```
***
## Best Practices
### 1. Resource Cleanup
Always disconnect MCP connections when done:
```python theme={null}
try:
# Use agent...
finally:
await mcp.disconnect()
```
### 2. Error Handling
Handle connection failures gracefully:
```python theme={null}
try:
await mcp.connect()
except MCPError as e:
logger.error(f"MCP connection failed: {e}")
# Fallback behavior
```
### 3. Server Names
Always provide `server_name` when using multiple servers:
```python theme={null}
MCPToolProvider(..., server_name="filesystem") # Good
MCPToolProvider(...) # Less clear with multiple servers
```
### 4. Auto Refresh
Use `auto_refresh=True` for hosted servers that may restart:
```python theme={null}
MCPToolProvider(url="https://...", auto_refresh=True)
```
### 5. Tool Discovery
Tools are automatically discovered and registered - no need to explicitly list them.
***
## Troubleshooting
### Connection Failures
If MCP connection fails:
* Check that the MCP server command is correct
* Verify network connectivity for HTTP/SSE transports
* Check API keys for authenticated servers
* Review logs for detailed error messages
### Tool Not Found
If an MCP tool is not available:
* Ensure MCP provider is connected: `mcp.is_connected()`
* Check tool name (may be prefixed with server name)
* Verify MCP server provides the expected tools
* Use `await mcp.discover_tools()` to list available tools
### Import Errors
If you see import errors:
* Install required dependencies: `pip install aiohttp`
* For stdio transport, ensure Node.js/npx is available
* Check that MCP server packages are installed
***
## Advanced Usage
### Custom Tool Prefixing
Control how tool names are prefixed:
```python theme={null}
mcp = MCPToolProvider(
command="...",
server_name="custom_name" # Tools will be: custom_name_tool_name
)
```
### Connection State Checking
Check if MCP provider is connected:
```python theme={null}
if mcp.is_connected():
tools = await mcp.discover_tools()
print(f"Available tools: {[t['name'] for t in tools]}")
```
### Manual Tool Registration
Register MCP tools manually into a ToolRegistry:
```python theme={null}
from laddr.core.tooling import register_mcp_tools
await register_mcp_tools(agent.tools, mcp_provider)
```
***
## Finding MCP Servers
Check out the [MCP Servers repository](https://github.com/modelcontextprotocol/servers) for a collection of available MCP servers:
* **Filesystem** - File and directory operations
* **Git** - Git repository operations
* **PostgreSQL** - Database queries
* **Brave Search** - Web search
* **And many more...**
***
## Next Steps
* [Tool Configuration](/guides/tools/tool-config) - Create custom tools
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
* [MCP Examples](/examples/mcp-examples) - See more MCP examples
* [Filesystem Agent Example](/examples/filesystem-agent) - Complete filesystem agent
# Tool Configuration
Source: https://laddr.agnetlabs.com/guides/tools/tool-config
Complete guide to creating and configuring tools for Laddr agents - decorators, schemas, patterns, and best practices
Learn how to create, configure, and use tools with Laddr agents.
***
## Basic Tool Structure
Tools are Python functions decorated with `@tool`:
```python theme={null}
from laddr import tool
@tool(
name="weather_lookup",
description="Fetches current weather for a given city",
parameters={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "City name"
},
"units": {
"type": "string",
"description": "Units (metric/imperial)",
"default": "metric"
}
},
"required": ["city"]
}
)
def weather_lookup(city: str, units: str = "metric") -> dict:
"""Tool docstring: fetch current weather."""
try:
result = get_weather(city, units)
return {"status": "success", "data": result}
except Exception as e:
return {"status": "error", "error": str(e)}
```
***
## The @tool Decorator
### Syntax
```python theme={null}
@tool(
name: str, # Unique tool identifier
description: str, # What the tool does
parameters: dict, # JSON Schema for inputs
trace: bool = True, # Enable tracing
trace_mask: list = [] # Fields to redact in traces
)
```
### Parameters
| Parameter | Type | Required | Description |
| ------------- | ------ | -------- | ------------------------------------------- |
| `name` | `str` | ✅ | Unique identifier used by agents |
| `description` | `str` | ✅ | Summary of tool's purpose |
| `parameters` | `dict` | ✅ | JSON Schema for tool inputs |
| `trace` | `bool` | ❌ | Enable or disable logging (default: `True`) |
| `trace_mask` | `list` | ❌ | Redact sensitive trace fields |
***
## JSON Schema Types
### String
```python theme={null}
parameters={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query"
}
},
"required": ["query"]
}
```
### Number
```python theme={null}
parameters={
"type": "object",
"properties": {
"count": {
"type": "integer",
"description": "Number of results",
"minimum": 1,
"maximum": 100
}
}
}
```
### Boolean
```python theme={null}
parameters={
"type": "object",
"properties": {
"include_metadata": {
"type": "boolean",
"description": "Include metadata in response"
}
}
}
```
### Array
```python theme={null}
parameters={
"type": "object",
"properties": {
"items": {
"type": "array",
"items": {"type": "string"},
"description": "List of items to process"
}
}
}
```
### Object (Nested)
```python theme={null}
parameters={
"type": "object",
"properties": {
"filters": {
"type": "object",
"properties": {
"min_price": {"type": "number"},
"max_price": {"type": "number"}
}
}
}
}
```
***
## Tool Patterns
### API Wrapper
```python theme={null}
import requests
from laddr import tool
@tool(
name="api_call",
description="Call external API",
parameters={
"type": "object",
"properties": {
"endpoint": {"type": "string"},
"method": {"type": "string", "enum": ["GET", "POST"]},
"data": {"type": "object"}
},
"required": ["endpoint"]
}
)
def api_call(endpoint: str, method: str = "GET", data: dict = None):
"""Make API call."""
try:
if method == "GET":
response = requests.get(endpoint)
else:
response = requests.post(endpoint, json=data)
response.raise_for_status()
return {"status": "success", "data": response.json()}
except Exception as e:
return {"status": "error", "error": str(e)}
```
### Data Transformer
```python theme={null}
from laddr import tool
@tool(
name="transform_data",
description="Transform data format",
parameters={
"type": "object",
"properties": {
"data": {"type": "array"},
"format": {"type": "string", "enum": ["json", "csv", "xml"]}
},
"required": ["data", "format"]
}
)
def transform_data(data: list, format: str) -> dict:
"""Transform data to specified format."""
if format == "json":
return {"format": "json", "data": data}
elif format == "csv":
# Convert to CSV
return {"format": "csv", "data": convert_to_csv(data)}
# ...
```
### File Operations
```python theme={null}
from laddr import tool
import os
@tool(
name="read_file",
description="Read file contents",
parameters={
"type": "object",
"properties": {
"file_path": {"type": "string"}
},
"required": ["file_path"]
}
)
def read_file(file_path: str) -> dict:
"""Read file from filesystem."""
try:
if not os.path.exists(file_path):
return {"status": "error", "error": "File not found"}
with open(file_path, "r") as f:
content = f.read()
return {"status": "success", "content": content}
except Exception as e:
return {"status": "error", "error": str(e)}
```
***
## Registering Tools with Agents
### Single Tool
```python theme={null}
from laddr import Agent
agent = Agent(
name="weather_agent",
tools=[weather_lookup],
# ... other config
)
```
### Multiple Tools
```python theme={null}
agent = Agent(
name="multi_tool_agent",
tools=[
weather_lookup,
api_call,
transform_data,
read_file
],
# ... other config
)
```
***
## Tracing Configuration
### Enable/Disable Tracing
```python theme={null}
@tool(
name="sensitive_tool",
description="Tool with sensitive data",
parameters={...},
trace=False # Disable tracing
)
def sensitive_tool(...):
pass
```
### Mask Sensitive Fields
```python theme={null}
@tool(
name="api_with_auth",
description="API call with authentication",
parameters={...},
trace_mask=["api_key", "password"] # Redact these fields
)
def api_with_auth(api_key: str, password: str, ...):
# api_key and password will be redacted in traces
pass
```
***
## Error Handling
### Return Error Status
```python theme={null}
@tool(name="risky_tool", ...)
def risky_tool(...):
try:
result = perform_operation()
return {"status": "success", "data": result}
except ValueError as e:
return {"status": "error", "error": f"Invalid input: {e}"}
except Exception as e:
return {"status": "error", "error": str(e)}
```
### Raise Exceptions
```python theme={null}
@tool(name="critical_tool", ...)
def critical_tool(...):
if not validate_input(...):
raise ValueError("Invalid input")
result = perform_operation()
return {"status": "success", "data": result}
```
Return error dictionaries for recoverable errors. Raise exceptions for critical failures.
***
## Testing Tools
### Manual Testing
```python theme={null}
# Test tool directly
result = weather_lookup(city="San Francisco", units="metric")
print(result)
# Test with invalid input
result = weather_lookup(city="")
print(result)
```
### Unit Testing
```python theme={null}
import pytest
from tools.weather_tools import weather_lookup
def test_weather_lookup_success():
result = weather_lookup(city="San Francisco")
assert result["status"] == "success"
assert "data" in result
def test_weather_lookup_invalid_city():
result = weather_lookup(city="")
assert result["status"] == "error"
```
### Integration Testing
```python theme={null}
async def test_tool_with_agent():
agent = Agent(
name="test_agent",
tools=[weather_lookup],
# ... config
)
result = await agent.process_task({
"query": "What's the weather in San Francisco?"
})
# Verify tool was called
assert "weather_lookup" in str(result)
```
***
## Best Practices
### 1. Clear Descriptions
```python theme={null}
# ✅ Good
@tool(
name="search_web",
description="Search the web for information on a given topic",
...
)
# ❌ Bad
@tool(
name="search",
description="Searches stuff",
...
)
```
### 2. Validate Inputs
```python theme={null}
@tool(name="safe_tool", ...)
def safe_tool(value: int):
if value < 0:
return {"status": "error", "error": "Value must be positive"}
# ... rest of implementation
```
### 3. Consistent Return Format
```python theme={null}
# ✅ Good - Consistent format
return {"status": "success", "data": result}
return {"status": "error", "error": "message"}
# ❌ Bad - Inconsistent
return result
return {"error": "message"}
return {"success": True, "result": result}
```
### 4. Handle Errors Gracefully
```python theme={null}
@tool(name="robust_tool", ...)
def robust_tool(...):
try:
result = perform_operation()
return {"status": "success", "data": result}
except SpecificError as e:
# Handle specific error
return {"status": "error", "error": str(e)}
except Exception as e:
# Log unexpected errors
logger.error(f"Unexpected error: {e}")
return {"status": "error", "error": "Internal error"}
```
***
## Advanced Patterns
### Async Tools
```python theme={null}
from laddr import tool
import aiohttp
@tool(name="async_api_call", ...)
async def async_api_call(url: str):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return {"status": "success", "data": await response.json()}
```
### Tool Composition
```python theme={null}
@tool(name="composite_tool", ...)
def composite_tool(...):
# Use other tools internally
result1 = helper_tool1(...)
result2 = helper_tool2(...)
return combine_results(result1, result2)
```
***
## Next Steps
* [MCP Integration](/guides/tools/mcp-integration) - Connect to MCP servers
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
* [System Tools](/guides/system-tools/custom-system-tools) - Custom system tools
* [Examples](/examples/filesystem-agent) - See tool examples
# Traces & Database
Source: https://laddr.agnetlabs.com/guides/traces
How to configure tracing, databases, and Langfuse in Laddr - SQLite, PostgreSQL, and Langfuse integration
# Traces & Database
Laddr supports flexible observability depending on the environment you're running in.\
This guide explains how to plug in each supported backend — SQLite, PostgreSQL, and Langfuse — and how Laddr automatically chooses the tracing mode.
***
## Overview
Laddr supports **three tracing modes**:
1. **SQLite Tracing (default)**\
Full local traces stored internally. Ideal for development.
2. **PostgreSQL Operational Mode**\
Used in production. Stores jobs, prompts, memory, batches — **but disables trace storage**.
3. **Langfuse Tracing (recommended for production)**\
External, scalable, full tracing UI + cost tracking.
The runtime automatically selects the best backend based on configuration.
***
## Backend Selection Logic
Laddr picks the tracing backend using the following priority:
> **1. Langfuse** (if enabled & credentials set)\
> **2. SQLite Internal Tracing** (if using SQLite)\
> **3. Disabled** (if using Postgres without Langfuse)
You can always verify which backend is active:
```bash theme={null}
curl http://localhost:8000/api/health
```
Expected example:
```json theme={null}
{
"components": {
"tracing": {
"enabled": true,
"backend": "langfuse"
}
}
}
```
***
## 1. Using SQLite (Local Development)
SQLite is the default mode and requires **no configuration**.
### Why SQLite?
* Zero setup
* Full trace storage
* Best for development/debugging
* Laddr dashboard can show all traces locally
### Enable
You don't need to set anything — Laddr defaults to:
```bash theme={null}
DATABASE_URL=sqlite:///./laddr.db
```
Or set your custom file:
```bash theme={null}
DATABASE_URL=sqlite:///./data/laddr.db
```
### Result
* Full tracing enabled automatically
* All traces visible in the `/traces` API and dashboard
***
## 2. Using PostgreSQL (Production Storage)
Use PostgreSQL for production deployments where multiple Laddr agents/services run.
### Why PostgreSQL?
* Durable storage for jobs, prompts, agent registry, batches, memory
* Supports horizontal scaling
* Suitable for multi-instance workloads
### Important:
> **Trace storage is automatically disabled when using Postgres.**
> Use Langfuse for production observability.
### Enable
```bash theme={null}
DB_BACKEND=postgres
DATABASE_URL=postgresql://user:password@host:5432/laddr
```
### Result
* Job/prompt/memory data stored in Postgres
* Internal trace events are **not** written
* Laddr expects Langfuse for observability
***
## 3. Enabling Langfuse (Recommended)
Langfuse provides the richest observability experience:
* Span tree visualization
* Prompt + token usage
* Cost tracking
* Filters & analytics
* Automatic LLM metadata aggregation
### Install
```bash theme={null}
pip install langfuse
```
### Configure
```bash theme={null}
LANGFUSE_ENABLED=true
LANGFUSE_PUBLIC_KEY=pk-...
LANGFUSE_SECRET_KEY=sk-...
LANGFUSE_HOST=https://cloud.langfuse.com
```
### Result
* Laddr stops writing traces to SQLite/Postgres
* All traces streamed to Langfuse
* Langfuse becomes the single tracing backend
***
## 4. Example Configurations
Below are realistic configurations for typical deployments.
***
### Local Development (SQLite + Tracing)
```bash theme={null}
DATABASE_URL=sqlite:///./laddr.db
DB_BACKEND=sqlite
ENABLE_TRACING=true
LANGFUSE_ENABLED=false
```
Outcome:
* Full trace logs stored in SQLite
* Inspect traces in dashboard or `/api/traces`
***
### Production Deployment (Postgres + Langfuse)
```bash theme={null}
DB_BACKEND=postgres
DATABASE_URL=postgresql://user:pass@host:5432/laddr
LANGFUSE_ENABLED=true
LANGFUSE_PUBLIC_KEY=pk-123
LANGFUSE_SECRET_KEY=sk-456
LANGFUSE_HOST=https://cloud.langfuse.com
ENABLE_TRACING=true # Langfuse handles tracing
```
Outcome:
* Reliable operational data in Postgres
* Full observability in Langfuse
* No duplicate trace writes
***
### Hybrid Cloud Environment (Local Tracing + Remote Postgres)
Useful when you're running a local Laddr node but using a shared Postgres cluster.
```bash theme={null}
DB_BACKEND=sqlite # Use SQLite for traces
DATABASE_URL=sqlite:///./local_traces.db
REMOTE_RESULT_DB_URL=postgresql://user:pass@host:5432/laddr
```
Outcome:
* Local tracing with SQLite
* Remote Postgres for persistent job/prompt data
***
## 5. Common Commands & Usage
### Check tracing backend
```bash theme={null}
curl http://localhost:8000/api/health
```
### Get traces (SQLite only)
```bash theme={null}
curl http://localhost:8000/api/traces?limit=100
curl http://localhost:8000/api/traces?job_id=
curl http://localhost:8000/api/traces/grouped
```
### View prompt execution traces (WebSocket)
```bash theme={null}
# Connect via WebSocket client
ws://localhost:8000/api/prompts//traces
```
### Check metrics
```bash theme={null}
curl http://localhost:8000/api/metrics
```
***
## 6. Troubleshooting
### Traces not appearing?
Check health:
```bash theme={null}
curl http://localhost:8000/api/health
```
Likely causes:
* You're using **Postgres** → internal traces are disabled
* Langfuse credentials missing
* `LANGFUSE_ENABLED=true` but SDK not installed
* `ENABLE_TRACING=false`
***
### Langfuse not receiving spans?
Check:
1. `pip install langfuse`
2. Valid public/secret keys
3. Host reachable
4. Runtime logs for connection issues
***
### SQLite issues?
* File permission errors in Linux
* Wrong directory path
* Conflicts when running multiple processes
***
## 7. Best Practices
### For Development
* Use **SQLite**
* Always keep internal tracing enabled
* Inspect traces in the dashboard frequently
* Use Langfuse selectively when debugging advanced workflows
### For Production
* Use **PostgreSQL**
* Never rely on internal DB for tracing
* Always enable **Langfuse**
* Configure retention policies
* Track LLM & tool call latency in Langfuse
### For Scaling
* Separate operational data (Postgres) from observability (Langfuse)
* Use container orchestration (Docker/Kubernetes)
* Distribute agents across workers but centralize your Postgres + Langfuse
***
## Next Steps
* [Storage & Artifacts](/guides/storage-and-artifacts) - Configure artifact storage
* [Scaling & Operations](/guides/scaling-and-ops) - Production deployment
* [Local Runtime](/guides/local-runtime) - Local development setup
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
# What is Laddr?
Source: https://laddr.agnetlabs.com/introduction/index
Scalable multi-agent intelligence framework for building and orchestrating intelligent agent systems
**Laddr** is an open-source multi-agent framework that helps developers build, orchestrate, and scale intelligent multi-agent systems with ease.
AI teams and developers around the world use Laddr to manage distributed agent systems, trace execution, and scale parallel tasks effortlessly.
***
## Why Laddr?
Building production-ready multi-agent systems is complex. You need:
* **Orchestration** - Coordinate multiple agents working together
* **Scalability** - Handle high throughput with parallel workers
* **Observability** - Track execution, debug issues, monitor performance
* **Flexibility** - Connect your own tools, APIs, and models
* **Reliability** - Handle failures, retries, and error recovery
Laddr provides all of this out of the box.
***
## Core Features
### Scalability
Queue-based architecture with parallel agent workers. Scale horizontally by adding more workers, or vertically by increasing worker capacity.
```bash theme={null}
# Scale workers dynamically
laddr scale researcher 5
laddr scale coordinator 3
```
### Observability
Docker-based dashboard with real-time traces, token usage, and system metrics. Track every agent execution, tool call, and LLM interaction.
```bash theme={null}
# Access dashboard
laddr run dev
# Open http://localhost:5173
```
### Extensibility
Connect your own tools, APIs, and external models with full control. Use MCP (Model Context Protocol), custom tools, or system tool overrides.
```python theme={null}
from laddr import Agent
from laddr.core.mcp_tools import MCPToolProvider
mcp = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-filesystem /path",
transport="stdio"
)
agent = Agent(
name="filesystem_agent",
tools=[mcp],
# ... config
)
```
### Configurability
Flexible PostgreSQL and MinIO/S3 storage options. Choose your queue backend (Redis, Kafka, or in-memory), LLM provider, and storage backend.
```bash theme={null}
# .env
QUEUE_BACKEND=redis
DB_BACKEND=postgresql
STORAGE_BACKEND=minio
LLM_PROVIDER=openai
```
### APIs
Exposed REST endpoints for jobs, traces, prompts, and system health. Integrate Laddr into your existing infrastructure.
```bash theme={null}
# Submit a job
curl -X POST http://localhost:8000/api/prompts \
-H "Content-Type: application/json" \
-d '{
"prompt_name": "researcher",
"inputs": {"query": "Latest AI trends"}
}'
```
***
## Architecture Overview
Laddr follows a distributed architecture pattern:
```mermaid theme={null}
graph TB
API[API Server] --> Queue[Message Queue
Redis/Kafka]
Queue --> Worker1[Agent Worker 1]
Queue --> Worker2[Agent Worker 2]
Queue --> Worker3[Agent Worker N]
Worker1 --> DB[(PostgreSQL)]
Worker2 --> DB
Worker3 --> DB
Worker1 --> Storage[(MinIO/S3)]
Worker2 --> Storage
Worker3 --> Storage
API --> Dashboard[Dashboard UI]
Dashboard --> DB
```
**Components:**
* **API Server** - REST API for submitting jobs and queries
* **Message Queue** - Task distribution (Redis, Kafka, or in-memory)
* **Agent Workers** - Execute agent tasks in parallel
* **Database** - Store traces, jobs, and metadata
* **Storage** - Store large artifacts and data
* **Dashboard** - Web UI for monitoring and debugging
***
## Quick Start
Get started with Laddr in under 60 seconds:
Run your first agent in 60 seconds
Set up Laddr on your machine
Explore Laddr's capabilities
Integrate with REST API
***
## Next Steps
* [Quickstart Guide](/introduction/quickstart) - Run your first agent
* [Installation](/getting-started/install) - Set up Laddr
* [Key Features](/introduction/key-features) - Learn what Laddr can do
* [First Agent](/getting-started/first-agent) - Build your first agent
# Key Features
Source: https://laddr.agnetlabs.com/introduction/key-features
Explore Laddr's core capabilities and features - scalability, observability, extensibility, and configurability
Laddr provides everything you need to build production-ready multi-agent systems.
***
## Scalability
Queue-based architecture with horizontal scaling support.
### Parallel Workers
Run multiple agent workers in parallel to handle high throughput:
```bash theme={null}
# Scale workers
laddr scale researcher 5
laddr scale coordinator 3
```
### Queue Backends
Choose your message queue backend:
* **Redis** - Fast, lightweight (development)
* **Kafka** - Durable, scalable (production)
* **Memory** - In-memory (testing)
```bash theme={null}
# .env
QUEUE_BACKEND=redis # or kafka, memory
```
### Load Distribution
Kafka automatically distributes tasks across workers using consumer groups:
```mermaid theme={null}
graph LR
Queue[Task Queue] --> W1[Worker 1]
Queue --> W2[Worker 2]
Queue --> W3[Worker 3]
Queue --> W4[Worker 4]
W1 --> Result[Results]
W2 --> Result
W3 --> Result
W4 --> Result
```
***
## Observability
Complete visibility into agent execution with traces, metrics, and dashboards.
### Real-Time Dashboard
Web-based dashboard at `http://localhost:5173`:
* **Agent Traces** - See every execution step
* **Token Usage** - Track LLM costs
* **System Metrics** - Monitor performance
* **Job History** - Review past executions
### Structured Traces
Every agent execution is traced:
```json theme={null}
{
"job_id": "abc-123",
"agent_name": "researcher",
"events": [
{"type": "task_start", "timestamp": "..."},
{"type": "llm_call", "tokens": 150},
{"type": "tool_call", "tool": "web_search"},
{"type": "task_complete", "result": "..."}
]
}
```
### Logging
Comprehensive logging at multiple levels:
```bash theme={null}
# View logs
laddr logs researcher --follow
laddr logs coordinator --tail 100
```
***
## Extensibility
Connect your own tools, APIs, and models with full control.
### Custom Tools
Create tools with the `@tool` decorator:
```python theme={null}
from laddr import tool
@tool(
name="weather_lookup",
description="Get weather for a city",
parameters={
"type": "object",
"properties": {
"city": {"type": "string"}
}
}
)
def weather_lookup(city: str):
return get_weather(city)
agent = Agent(
name="weather_agent",
tools=[weather_lookup]
)
```
### MCP Integration
Connect to Model Context Protocol servers:
```python theme={null}
from laddr.core.mcp_tools import MCPToolProvider
mcp = MCPToolProvider(
command="npx -y @modelcontextprotocol/server-filesystem /path",
transport="stdio"
)
agent = Agent(
name="filesystem_agent",
tools=[mcp]
)
```
### System Tool Overrides
Customize delegation and storage behavior:
```python theme={null}
from laddr import override_system_tool, TaskDelegationTool
@override_system_tool("system_delegate_task")
async def custom_delegate(
agent_name: str,
task_description: str,
task: str,
_message_bus=None,
_artifact_storage=None,
_agent=None
):
# Add custom logic
logger.info(f"Delegating to {agent_name}")
# Reuse base functionality
tool = TaskDelegationTool(_message_bus, _artifact_storage, _agent)
return await tool.delegate_task(...)
```
***
## Configurability
Flexible configuration for every component.
### Storage Backends
Choose your storage backend:
* **MinIO** - S3-compatible object storage
* **S3** - AWS S3
* **Local** - File system (development)
```bash theme={null}
# .env
STORAGE_BACKEND=minio
MINIO_ENDPOINT=localhost:9000
MINIO_ACCESS_KEY=minioadmin
MINIO_SECRET_KEY=minioadmin
```
### Database Options
* **PostgreSQL** - Production database
* **SQLite** - Local development
```bash theme={null}
# .env
DB_BACKEND=postgresql
DATABASE_URL=postgresql://user:pass@localhost:5432/laddr
```
### LLM Providers
Support for multiple LLM providers:
* **OpenAI** - GPT-4, GPT-3.5
* **Anthropic** - Claude
* **Google** - Gemini
* **Ollama** - Local models
```python theme={null}
from laddr.llms import openai, gemini, ollama
# OpenAI
agent = Agent(..., llm=openai(model="gpt-4o-mini"))
# Gemini
agent = Agent(..., llm=gemini(model="gemini-2.5-flash"))
# Ollama (local)
agent = Agent(..., llm=ollama(model="llama3.2:latest"))
```
***
## APIs
REST API for integration with your systems.
### Submit Jobs
```http theme={null}
POST /api/prompts
Content-Type: application/json
{
"prompt_name": "researcher",
"inputs": {"query": "Latest AI trends"}
}
```
### Get Results
```http theme={null}
GET /api/prompts/{prompt_id}
```
### Health Check
```http theme={null}
GET /api/health
```
See the [API Reference](/reference/api) for complete documentation.
***
## Agent Orchestration
Coordinate multiple agents working together.
### Task Delegation
Agents can delegate tasks to other agents:
```python theme={null}
# In coordinator agent
result = await self.delegate_task(
agent_name="researcher",
task_description="Research AI trends",
task="research",
task_data={"topic": "AI trends"}
)
```
### Parallel Execution
Run multiple tasks in parallel:
```python theme={null}
results = await self.delegate_parallel(
agent_name="analyzer",
tasks=[
{"task": "analyze", "task_data": {"data": data1}},
{"task": "analyze", "task_data": {"data": data2}},
{"task": "analyze", "task_data": {"data": data3}}
]
)
```
### Workflow Patterns
```mermaid theme={null}
graph TD
Start[Start] --> Coordinator[Coordinator Agent]
Coordinator --> Researcher[Researcher Agent]
Coordinator --> Analyzer[Analyzer Agent]
Researcher --> Writer[Writer Agent]
Analyzer --> Writer
Writer --> End[End]
```
***
## Error Handling
Built-in retry logic and error recovery.
### Automatic Retries
```python theme={null}
agent = Agent(
name="researcher",
max_retries=3, # Retry failed tasks
timeout=60 # Timeout in seconds
)
```
### Error Tracking
All errors are logged and traced:
```json theme={null}
{
"event_type": "tool_error",
"tool": "web_search",
"error": "API rate limit exceeded",
"retry_count": 1
}
```
***
## Next Steps
* [Installation](/getting-started/install) - Set up Laddr
* [First Agent](/getting-started/first-agent) - Build your first agent
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
* [Scaling & Operations](/guides/scaling-and-ops) - Production deployment
# Quickstart
Source: https://laddr.agnetlabs.com/introduction/quickstart
Get started with Laddr in under 60 seconds - run your first agent
Get Laddr running and execute your first agent in under 60 seconds.
***
## Prerequisites
* Python 3.9+
* pip installed
* (Optional) Docker for full stack
***
## Step 1: Install Laddr
```bash theme={null}
pip install laddr
```
Verify installation:
```bash theme={null}
laddr --version
```
***
## Step 2: Initialize a Project
```bash theme={null}
laddr init my_agent_system
cd my_agent_system
```
This creates a project structure with:
* `agents/` - Your agent definitions
* `workers/` - Worker scripts
* `docker-compose.yml` - Docker configuration
* `main.py` - Runner script
***
## Step 3: Set API Keys
Create a `.env` file in your project root:
```bash theme={null}
# .env
GEMINI_API_KEY=your_gemini_api_key
SERPER_API_KEY=your_serper_api_key
```
You can use any LLM provider. Gemini is used in this example, but OpenAI, Anthropic, and Ollama are also supported.
***
## Step 4: Add an Agent
```bash theme={null}
laddr add agent researcher \\
--role "Research Specialist" \\
--goal "Find and analyze information" \\
--llm-model gemini-2.5-flash
```
***
## Step 5: Add a Tool
```bash theme={null}
laddr add tool web_search \\
--agent researcher \\
--description "Search the web for information"
```
***
## Step 6: Run the Agent
### Option A: Local Run (Fastest)
```bash theme={null}
laddr run-local researcher --input '{"query": "What is Laddr?"}'
```
### Option B: Full Stack (Docker)
```bash theme={null}
laddr run dev -d
docker compose up -d
```
Then submit a job:
```bash theme={null}
laddr prompt run researcher --input query="What is Laddr?"
```
***
## Expected Output
You should see:
```
Starting researcher agent...
Processing query: What is Laddr?
Using tool: web_search
Found results: [...]
Agent response: Laddr is an open-source multi-agent framework...
Task completed successfully
```
***
## What's Next?
Learn to create custom agents
Use the web UI to monitor agents
Configure agents in detail
Create custom tools
***
## Troubleshooting
### Installation Issues
If `laddr` command not found:
```bash theme={null}
# Add to PATH or use python -m
python -m laddr --version
```
### API Key Errors
Ensure your `.env` file is in the project root and keys are valid.
### Docker Issues
If Docker commands fail, ensure Docker is running:
```bash theme={null}
docker ps
```
***
## Next Steps
* [Installation Guide](/getting-started/install) - Detailed setup instructions
* [First Agent Tutorial](/getting-started/first-agent) - Build a custom agent
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
* [Tool Development](/guides/tools/tool-config) - Create custom tools
# API Reference
Source: https://laddr.agnetlabs.com/reference/api
Complete REST API documentation for Laddr - endpoints, authentication, WebSockets, and examples
Complete documentation of all Laddr API endpoints.
***
## Base URL
All API requests should be made to:
**[http://localhost:8000](http://localhost:8000)**
For production deployments, replace `localhost` with your server's hostname or IP address.
***
## Authentication
API key authentication is optional. If `LADDR_API_KEY` environment variable is set, all endpoints require authentication.
**Header Format:**
```http theme={null}
X-API-Key: your-api-key
```
**Alternative (Bearer Token):**
```http theme={null}
Authorization: Bearer your-api-key
```
**WebSocket Authentication:**
* Query parameter: `?api_key=your-api-key`
* Or use `X-API-Key` header
* Or use `Authorization: Bearer your-api-key` header
If `LADDR_API_KEY` is not set, authentication is disabled (no-op).
***
## Health
Check the system's health status.
```http theme={null}
GET /api/health
```
**Response:**
```json theme={null}
{
"status": "ok",
"version": "0.8.6",
"components": {
"database": "SQLite",
"storage": "MinIO",
"message_bus": "REDIS",
"tracing": {
"enabled": true,
"backend": "database"
}
}
}
```
**Tracing Backend Values:**
* `"database"` - SQLite-based internal tracing
* `"langfuse"` - Langfuse external tracing
* `"disabled"` - Tracing not available
***
## Prompts (Preferred)
The prompts API is the recommended way to submit tasks to agents.
### Create Prompt
Submit a new prompt execution.
```http theme={null}
POST /api/prompts
Content-Type: application/json
{
"prompt_name": "researcher",
"inputs": {
"query": "What is Laddr?"
},
"mode": "single",
"agents": null
}
```
**Request Body:**
* `prompt_name` (required) - Name of the agent to execute
* `inputs` (required) - Input data for the agent
* `mode` (optional) - Execution mode: `"single"` (default) or `"sequential"`
* `agents` (optional) - For sequential mode: ordered list of agent names to run in sequence
**Response:**
```json theme={null}
{
"prompt_id": "abc-123-def-456",
"status": "running",
"agent": "researcher",
"mode": "single",
"agents": ["researcher"]
}
```
**Sequential Mode Example:**
```http theme={null}
POST /api/prompts
Content-Type: application/json
{
"prompt_name": "coordinator",
"inputs": {"topic": "AI research"},
"mode": "sequential",
"agents": ["researcher", "writer", "reviewer"]
}
```
This runs agents in order, piping output from one to the next.
### Get Prompt
Retrieve details of a specific prompt execution.
```http theme={null}
GET /api/prompts/{prompt_id}
```
**Response:**
```json theme={null}
{
"prompt_id": "abc-123-def-456",
"prompt_name": "researcher",
"status": "completed",
"inputs": {"query": "What is Laddr?"},
"outputs": {"answer": "Laddr is..."},
"created_at": "2024-01-01T00:00:00Z",
"completed_at": "2024-01-01T00:00:05Z",
"token_usage": {
"prompt_tokens": 150,
"completion_tokens": 75,
"total_tokens": 225,
"by_model": [
{
"provider": "openai",
"model": "gpt-4",
"prompt_tokens": 150,
"completion_tokens": 75,
"total_tokens": 225,
"calls": 1
}
]
}
}
```
### List Prompts
List all prompt executions with pagination.
```http theme={null}
GET /api/prompts?limit=50
```
**Query Parameters:**
* `limit` - Maximum number of results (default: 50)
**Response:**
```json theme={null}
{
"prompts": [
{
"prompt_id": "abc-123",
"prompt_name": "researcher",
"status": "completed",
"created_at": "2024-01-01T00:00:00Z",
"completed_at": "2024-01-01T00:00:05Z"
}
],
"limit": 50
}
```
### Cancel Prompt
Cancel a running prompt execution.
```http theme={null}
POST /api/prompts/{prompt_id}/cancel
```
**Response:**
```json theme={null}
{
"ok": true,
"prompt_id": "abc-123-def-456",
"status": "canceled"
}
```
***
## Jobs (Legacy)
The jobs API is maintained for backward compatibility. Use the prompts API for new integrations.
### Submit Job
Submit a job using the legacy endpoint.
```http theme={null}
POST /api/jobs
Content-Type: application/json
{
"pipeline_name": "coordinator",
"inputs": {"topic": "Latest AI agents"}
}
```
**Response:**
```json theme={null}
{
"job_id": "abc-123-def-456",
"status": "success",
"result": {"output": "..."},
"error": null,
"duration_ms": 1234,
"agent": "coordinator"
}
```
### Get Job
Retrieve a specific job by ID.
```http theme={null}
GET /api/jobs/{job_id}
```
**Response:**
```json theme={null}
{
"job_id": "abc-123-def-456",
"status": "completed",
"pipeline_name": "coordinator",
"inputs": {"topic": "Latest AI agents"},
"outputs": {"result": "..."},
"created_at": "2024-01-01T00:00:00Z",
"completed_at": "2024-01-01T00:00:05Z",
"token_usage": {
"prompt_tokens": 1000,
"completion_tokens": 500,
"total_tokens": 1500,
"by_model": [...]
}
}
```
### List Jobs
List all jobs with pagination support.
```http theme={null}
GET /api/jobs?limit=50&offset=0
```
### Replay Job
Replay a previous job execution.
```http theme={null}
POST /api/jobs/{job_id}/replay
Content-Type: application/json
{
"reexecute": false
}
```
**Request Body:**
* `reexecute` - If `true`, re-run the job. If `false`, return stored result.
***
## Batches
Batch operations allow you to submit multiple tasks to an agent in parallel. Each task gets its own unique `job_id` and `trace_id`, but all tasks are grouped under a `batch_id` for tracking.
### Submit Batch Tasks
Submit multiple tasks to an agent's queue in parallel.
```http theme={null}
POST /api/agents/{agent_name}/batch
Content-Type: application/json
{
"tasks": [
{"query": "What is Python?"},
{"query": "What is JavaScript?"},
{"query": "What is Rust?"}
],
"wait": false,
"batch_id": null
}
```
**Request Body:**
* `tasks` (required) - List of task payloads to execute in parallel
* `wait` (optional) - If `true`, wait for all responses before returning (default: `false`)
* `batch_id` (optional) - Existing batch ID to add tasks to, or `null` to create new batch
**Response (non-blocking):**
```json theme={null}
{
"batch_id": "batch-abc-123",
"agent_name": "researcher",
"status": "submitted",
"task_count": 3,
"task_ids": ["task-1", "task-2", "task-3"],
"job_ids": ["job-1", "job-2", "job-3"],
"trace_ids": ["trace-1", "trace-2", "trace-3"]
}
```
**Response (blocking, wait=true):**
```json theme={null}
{
"batch_id": "batch-abc-123",
"agent_name": "researcher",
"status": "completed",
"task_count": 3,
"task_ids": ["task-1", "task-2", "task-3"],
"job_ids": ["job-1", "job-2", "job-3"],
"trace_ids": ["trace-1", "trace-2", "trace-3"],
"results": [
{
"task_id": "task-1",
"response": {"status": "success", "result": "..."}
},
{
"task_id": "task-2",
"response": {"status": "success", "result": "..."}
},
{
"task_id": "task-3",
"response": {"status": "success", "result": "..."}
}
]
}
```
### Add Tasks to Batch
Add more tasks to an existing batch (useful for adding aggregator tasks after evaluator workers complete).
```http theme={null}
POST /api/batches/{batch_id}/add-tasks
Content-Type: application/json
{
"agent_name": "aggregator",
"tasks": [
{"batch_id": "batch-abc-123", "operation": "summarize"}
],
"wait": false
}
```
**Request Body:**
* `agent_name` (required) - Agent to run the new tasks
* `tasks` (required) - List of task payloads to add
* `wait` (optional) - If `true`, wait for responses (default: `false`)
**Response:**
```json theme={null}
{
"batch_id": "batch-abc-123",
"status": "running",
"added_job_ids": ["job-4"],
"added_trace_ids": ["trace-4"],
"added_task_ids": ["task-4"],
"total_tasks": 4,
"total_job_ids": 4
}
```
### Get Batch
Retrieve batch metadata and status.
```http theme={null}
GET /api/batches/{batch_id}
```
**Response:**
```json theme={null}
{
"batch_id": "batch-abc-123",
"agent_name": "researcher",
"status": "completed",
"task_count": 3,
"job_ids": ["job-1", "job-2", "job-3"],
"task_ids": ["task-1", "task-2", "task-3"],
"inputs": {"tasks": [...]},
"outputs": {
"results": {
"job-1": {"status": "success", "response": {...}},
"job-2": {"status": "success", "response": {...}},
"job-3": {"status": "success", "response": {...}}
},
"summary": {
"total_expected": 3,
"recorded": 3,
"succeeded": 3,
"failed": 0
}
},
"created_at": "2024-01-01T00:00:00Z",
"completed_at": "2024-01-01T00:00:05Z"
}
```
### List Batches
List recent batch operations.
```http theme={null}
GET /api/batches?limit=50
```
**Query Parameters:**
* `limit` - Maximum number of batches to return (default: 50)
**Response:**
```json theme={null}
{
"batches": [
{
"batch_id": "batch-abc-123",
"agent_name": "researcher",
"status": "completed",
"task_count": 3,
"job_ids": ["job-1", "job-2", "job-3"],
"task_ids": ["task-1", "task-2", "task-3"],
"created_at": "2024-01-01T00:00:00Z",
"completed_at": "2024-01-01T00:00:05Z"
}
],
"limit": 50
}
```
***
## Agents
### List Agents
List all registered agents with metadata.
```http theme={null}
GET /api/agents
```
**Response:**
```json theme={null}
{
"agents": [
{
"name": "researcher",
"role": "Research Assistant",
"goal": "Conduct research on given topics",
"status": "active",
"tools": ["web_search", "read_document"],
"last_seen": "2024-01-01T00:00:00Z",
"trace_count": 150,
"last_executed": "2024-01-01T00:00:00Z"
}
]
}
```
### Chat with Agent
Send a message to an agent and optionally wait for response.
```http theme={null}
GET /api/agents/{agent_name}/chat?message=Hello&wait=true&timeout=30
```
**Query Parameters:**
* `message` (required) - Message to send to the agent
* `wait` (optional) - If `true`, wait for response (default: `true`)
* `timeout` (optional) - Timeout in seconds when waiting (default: 30)
**Response (wait=true):**
```json theme={null}
{
"task_id": "task-abc-123",
"status": "completed",
"response": {"message": "Hello! How can I help you?"}
}
```
**Response (wait=false):**
```json theme={null}
{
"task_id": "task-abc-123",
"status": "submitted"
}
```
### Get Agent Tools
Get detailed tool information for a specific agent.
```http theme={null}
GET /api/agents/{agent_name}/tools
```
**Response:**
```json theme={null}
{
"agent": "researcher",
"tools": [
{
"name": "web_search",
"description": "Search the web for information",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query"
}
},
"required": ["query"]
}
}
]
}
```
***
## Traces
Traces provide observability into agent execution, tool calls, and LLM interactions.
### List Traces
List trace events with optional filters.
```http theme={null}
GET /api/traces?job_id=abc-123&agent_name=researcher&limit=100
```
**Query Parameters:**
* `job_id` (optional) - Filter traces by job ID
* `agent_name` (optional) - Filter traces by agent name
* `limit` (optional) - Maximum number of traces to return (default: 100)
**Response:**
```json theme={null}
{
"traces": [
{
"id": 1,
"job_id": "abc-123",
"agent_name": "researcher",
"event_type": "task_start",
"parent_id": null,
"payload": {"task": "..."},
"timestamp": "2024-01-01T00:00:00Z"
},
{
"id": 2,
"job_id": "abc-123",
"agent_name": "researcher",
"event_type": "tool_call",
"parent_id": 1,
"payload": {
"tool_name": "web_search",
"parameters": {"query": "..."},
"result": {"status": "success"}
},
"timestamp": "2024-01-01T00:00:01Z"
}
]
}
```
### Get Grouped Traces
Get traces grouped by job\_id, showing complete multi-agent runs together.
```http theme={null}
GET /api/traces/grouped?limit=50
```
**Query Parameters:**
* `limit` (optional) - Maximum number of job groups to return (default: 50)
**Response:**
```json theme={null}
{
"grouped_traces": [
{
"job_id": "abc-123",
"trace_count": 15,
"agents": ["coordinator", "researcher"],
"start_time": "2024-01-01T00:00:00Z",
"end_time": "2024-01-01T00:00:05Z",
"traces": [...]
}
]
}
```
### Get Trace
Get a single trace event by ID with full payload.
```http theme={null}
GET /api/traces/{trace_id}
```
**Response:**
```json theme={null}
{
"id": 1,
"job_id": "abc-123",
"agent_name": "researcher",
"event_type": "tool_call",
"payload": {
"tool_name": "web_search",
"parameters": {"query": "Python async"},
"result": {"status": "success", "data": "..."},
"duration_ms": 245
},
"timestamp": "2024-01-01T00:00:01Z"
}
```
***
## Metrics
### Get Metrics
Get aggregated system metrics.
```http theme={null}
GET /api/metrics
```
**Response:**
```json theme={null}
{
"total_jobs": 150,
"completed_jobs": 140,
"failed_jobs": 10,
"avg_latency_ms": 1250,
"active_agents_count": 5,
"tool_calls": 500,
"cache_hits": 50,
"total_tokens": 100000,
"timestamp": "2024-01-01T00:00:00Z"
}
```
***
## Responses
### Get Resolved Response
Resolve a task response. If the response was offloaded to storage (MinIO/S3), this endpoint fetches the full payload.
```http theme={null}
GET /api/responses/{task_id}/resolved
```
**Response (inline):**
```json theme={null}
{
"task_id": "task-abc-123",
"offloaded": false,
"pointer": null,
"data": {"status": "success", "result": "..."}
}
```
**Response (offloaded):**
```json theme={null}
{
"task_id": "task-abc-123",
"offloaded": true,
"pointer": {
"bucket": "laddr",
"key": "responses/task-abc-123.json",
"size_bytes": 1024000
},
"data": {"status": "success", "result": "..."}
}
```
***
## Container Logs
### List Containers
List all Docker containers (project-agnostic).
```http theme={null}
GET /api/logs/containers
```
**Response:**
```json theme={null}
{
"containers": [
{
"id": "abc123def456",
"name": "laddr-api",
"service_name": "api",
"project_name": "laddr",
"type": "api",
"status": "running",
"image": "laddr:latest",
"created": "2024-01-01T00:00:00Z"
}
],
"total": 5
}
```
**Container Types:**
* `api` - API server containers
* `worker` - Agent worker containers
* `infrastructure` - Database, Redis, MinIO, etc.
* `other` - Other containers
### Get Container Logs
Get logs from a specific container.
```http theme={null}
GET /api/logs/containers/{container_name}?tail=100&since=5m×tamps=true
```
**Query Parameters:**
* `tail` (optional) - Number of lines to return (default: 100)
* `since` (optional) - Only logs since this timestamp (e.g., "5m", "1h", or ISO8601)
* `timestamps` (optional) - Include timestamps in logs (default: `true`)
**Response:**
```json theme={null}
{
"container": "laddr-api",
"container_id": "abc123def456",
"status": "running",
"logs": [
{
"timestamp": "2024-01-01T00:00:00.000000000Z",
"message": "Laddr API server started"
}
],
"total": 100
}
```
***
## WebSockets
### Prompt Traces
Stream live trace events for a specific prompt execution.
```http theme={null}
WS /ws/prompts/{prompt_id}
```
**Message Format:**
```json theme={null}
{
"type": "traces",
"data": {
"spans": [
{
"id": 1,
"name": "researcher",
"type": "agent",
"start_time": "2024-01-01T00:00:00Z",
"agent": "researcher",
"event_type": "task_start",
"input": {"query": "..."},
"output": null,
"metadata": {...},
"children": [
{
"id": 2,
"name": "web_search",
"type": "tool",
"input": {"query": "..."},
"output": {"result": "..."},
"children": []
}
]
}
],
"count": 15
}
}
```
**Completion Event:**
```json theme={null}
{
"type": "complete",
"data": {
"status": "completed",
"outputs": {"result": "..."},
"error": null,
"spans": [...]
}
}
```
### Batch Traces
Stream live trace events for a batch operation (all job\_ids in the batch).
```http theme={null}
WS /ws/batches/{batch_id}
```
**Message Format:**
Same as prompt traces, but includes traces from all `job_ids` in the batch, grouped by `job_id`.
### Container Logs
Stream container logs in real-time.
```http theme={null}
WS /ws/logs/{container_name}
```
**Message Format:**
```json theme={null}
{
"type": "log",
"data": {
"timestamp": "2024-01-01T00:00:00.000000000Z",
"message": "Laddr API server started"
}
}
```
### Events
Stream real-time system events (throttled).
```http theme={null}
WS /ws/events
```
**Message Format:**
```json theme={null}
{
"type": "trace",
"data": {
"job_id": "abc-123",
"agent_name": "researcher",
"event_type": "tool_call",
"timestamp": "2024-01-01T00:00:00Z"
}
}
```
**Batch Events:**
```json theme={null}
{
"type": "batch",
"events": [
{"type": "trace", "data": {...}},
{"type": "trace", "data": {...}}
]
}
```
***
## Error Responses
All endpoints may return error responses in the following format:
```json theme={null}
{
"detail": "Error message"
}
```
**Common Status Codes:**
* `200` - Success
* `400` - Bad Request
* `401` - Unauthorized (invalid or missing API key)
* `404` - Not Found
* `500` - Internal Server Error
* `502` - Bad Gateway (storage fetch failed)
* `503` - Service Unavailable (Docker SDK not available)
***
## Rate Limiting
Currently, there are no rate limits. For production deployments, consider implementing rate limiting.
***
## Event Types
Trace events use the following `event_type` values:
* `task_start` - Agent task execution begins
* `task_complete` - Agent task execution completes
* `tool_call` - Tool invocation with parameters and results
* `llm_usage` - LLM API call with token usage
* `cache_hit` - Cached result used
* `delegation` - Task delegation to another agent
* `error` - Error occurrence with stack trace
* `task_cancel_requested` - Task cancellation requested
## Next Steps
* [Traces and Database](/traces-and-database) - Understanding traces, SQLite, Langfuse, and Postgres
* [CLI Reference](/reference/cli) - Command-line interface
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
* [Getting Started](/getting-started/install) - Setup guide
# CLI Reference
Source: https://laddr.agnetlabs.com/reference/cli
Complete reference for all Laddr CLI commands - installation, agent management, running, and monitoring
# Laddr CLI Reference
Complete reference for all Laddr Command Line Interface (CLI) commands, options, and usage examples.
***
## Overview
The Laddr CLI provides commands for:
* **Project scaffolding** - Initialize complete project structures
* **Agent management** - Add, configure, and run agents
* **Docker orchestration** - Manage containerized environments
* **Observability** - Built-in tracing, metrics, and diagnostics
* **Development tools** - Hot reload, logging, and debugging
***
## Installation
```bash theme={null}
# Install Laddr
pip install laddr
# Verify installation
laddr --version
# Get help
laddr --help
```
***
## Global Options
Available for all commands:
* `-h, --help` - Show help message and exit
* `--version` - Show version and exit
* `--debug` - Enable debug mode with full stack traces
***
## Project Setup
### init
Initialize a new Laddr project.
```bash theme={null}
laddr init [PROJECT_NAME] [OPTIONS]
```
**Options:**
* `PROJECT_NAME` - Name of the project (interactive if omitted)
* `--path` - Path to create project in (default: current directory)
**Examples:**
```bash theme={null}
# Interactive mode
laddr init
# Specify project name
laddr init myproject
# Create in specific directory
laddr init myproject --path /path/to/projects
```
***
## Agent Management
### add agent
Add a new agent to the project.
```bash theme={null}
laddr add agent AGENT_NAME [OPTIONS]
```
**Options:**
* `--role` - Agent role
* `--goal` - Agent goal
* `--backstory` - Agent backstory (optional)
* `--llm-provider` - LLM provider (default: gemini)
* `--llm-model` - LLM model (default: gemini-2.5-flash)
**Examples:**
```bash theme={null}
# Interactive mode
laddr add agent summarizer
# Fully specified
laddr add agent summarizer \\
--role "Data Summarizer" \\
--goal "Analyze and summarize data"
```
### add tool
Add a new tool to an agent.
```bash theme={null}
laddr add tool TOOL_NAME [OPTIONS]
```
**Options:**
* `--agent` - Agent to attach tool to
* `--description` - Tool description
**Examples:**
```bash theme={null}
# Interactive mode
laddr add tool calculator
# Specify agent
laddr add tool scraper --agent researcher
```
***
## Running & Execution
### run dev
Run the Laddr development environment.
```bash theme={null}
laddr run dev [OPTIONS]
```
**Options:**
* `--build` - Force rebuild Docker images
* `--detach, -d` - Run in detached mode (background)
**Examples:**
```bash theme={null}
# Start with live logs
laddr run dev
# Start and run in background
laddr run dev --detach
# Force rebuild images
laddr run dev --build
```
### run agent
Run a single agent locally.
```bash theme={null}
laddr run agent AGENT_NAME [OPTIONS]
```
**Options:**
* `--inputs` - JSON dictionary of inputs (default: empty object)
**Examples:**
```bash theme={null}
# Run with query
laddr run agent researcher --inputs '{"query":"What is Laddr?"}'
# Run with no inputs
laddr run agent coordinator
```
### run-local
Run an agent locally using in-memory components.
```bash theme={null}
laddr run-local AGENT [OPTIONS]
```
**Options:**
* `--input` - JSON input to the agent (default: empty object)
**Examples:**
```bash theme={null}
# Run with no input
laddr run-local researcher
# Run with input
laddr run-local researcher --input '{"query":"AI trends"}'
```
### prompt run
Execute a prompt with an agent.
```bash theme={null}
laddr prompt run PROMPT_NAME [OPTIONS]
```
**Options:**
* `--input, -i` - Input as KEY=VALUE (repeatable)
* `--json, -j` - Input as JSON string or @file.json
* `--wait/--no-wait` - Wait for completion (default: true)
* `--timeout` - Timeout in seconds (default: 60)
**Examples:**
```bash theme={null}
# Simple input
laddr prompt run researcher --input query="What is Laddr?"
# JSON input
laddr prompt run processor --json '{"files": ["data.csv"]}'
# From file
laddr prompt run researcher --json @research-query.json
```
### prompt list
List recent prompt executions.
```bash theme={null}
laddr prompt list [OPTIONS]
```
**Options:**
* `--limit` - Maximum number to show (default: 20)
***
## Monitoring & Debugging
### logs
View logs for an agent worker.
```bash theme={null}
laddr logs AGENT_NAME [OPTIONS]
```
**Options:**
* `--follow, -f` - Follow log output (live tail)
* `--tail` - Number of lines to show from end
**Examples:**
```bash theme={null}
# View all logs
laddr logs researcher
# Follow logs
laddr logs researcher --follow
# Show last 100 lines
laddr logs researcher --tail 100
```
### ps
Show status of all services and workers.
```bash theme={null}
laddr ps [OPTIONS]
```
**Options:**
* `--format, -f` - Output format (table or json)
**Examples:**
```bash theme={null}
# Show status (table format)
laddr ps
# Show status (JSON format)
laddr ps --format json
```
### check
Run diagnostic checks for the Laddr runtime.
```bash theme={null}
laddr check [OPTIONS]
```
**Options:**
* `--output` - Path to write JSON report (default: diagnostic\_report.json)
**Examples:**
```bash theme={null}
# Run diagnostics
laddr check
# Save to custom path
laddr check --output reports/diagnostic.json
```
### infra
Show infrastructure configuration and status.
```bash theme={null}
laddr infra [OPTIONS]
```
**Options:**
* `--show-config/--no-show-config` - Show configuration values (default: true)
***
## Service Management
### scale
Scale an agent worker to N replicas.
```bash theme={null}
laddr scale AGENT_NAME REPLICAS
```
**Examples:**
```bash theme={null}
# Scale to 3 replicas
laddr scale researcher 3
# Scale to 1 replica
laddr scale researcher 1
# Stop all replicas
laddr scale researcher 0
```
### stop
Stop all Laddr services.
```bash theme={null}
laddr stop [OPTIONS]
```
**Options:**
* `--volumes, -v` - Remove named volumes (warning: data loss)
**Examples:**
```bash theme={null}
# Stop services (preserve data)
laddr stop
# Stop services and remove volumes
laddr stop --volumes
```
***
## Next Steps
* [API Reference](/reference/api) - REST API documentation
* [Getting Started](/getting-started/install) - Setup guide
* [Agent Configuration](/guides/agents/agent-config) - Configure agents
# Agent Schema
Source: https://laddr.agnetlabs.com/reference/schemas/agent-schema
Complete schema reference for Laddr agent configuration - required fields, optional fields, and examples
Complete reference for the Laddr agent configuration schema.
***
## Agent Configuration
```python theme={null}
from laddr import Agent
agent = Agent(
name="researcher",
role="Web Research Specialist",
goal="Search the web and summarize findings",
backstory="You are a research-focused agent...",
llm=llm_config,
tools=[tool1, tool2],
max_retries=1,
max_iterations=3,
max_tool_calls=10,
timeout=300,
trace_enabled=True,
instructions="Custom instructions...",
is_coordinator=False
)
```
***
## Required Fields
| Field | Type | Description |
| ----------- | ----- | ---------------------------- |
| `name` | `str` | Unique agent identifier |
| `role` | `str` | Agent's role description |
| `goal` | `str` | Primary objective or task |
| `backstory` | `str` | Context that shapes behavior |
| `llm` | `LLM` | LLM configuration object |
***
## Optional Fields
| Field | Type | Default | Description |
| ---------------- | ------ | ------- | ---------------------------------- |
| `tools` | `list` | `[]` | List of tool functions |
| `max_retries` | `int` | `1` | Number of retries for failed tasks |
| `max_iterations` | `int` | `3` | Maximum reasoning loops |
| `max_tool_calls` | `int` | `10` | Maximum tool calls per iteration |
| `timeout` | `int` | `300` | Task timeout in seconds |
| `trace_enabled` | `bool` | `True` | Enable execution tracing |
| `instructions` | `str` | `None` | Custom instructions for behavior |
| `is_coordinator` | `bool` | `False` | Enable delegation tools |
***
## LLM Configuration
### OpenAI
```python theme={null}
from laddr.llms import openai
llm = openai(
model="gpt-4o-mini",
temperature=0.0,
max_tokens=2000
)
```
### Gemini
```python theme={null}
from laddr.llms import gemini
llm = gemini(
model="gemini-2.5-flash",
temperature=0.0
)
```
### Anthropic
```python theme={null}
from laddr.llms import anthropic
llm = anthropic(
model="claude-3-5-sonnet-20241022",
temperature=0.0
)
```
### Ollama
```python theme={null}
from laddr.llms import ollama
llm = ollama(
model="llama3.2:latest",
base_url="http://localhost:11434"
)
```
***
## Example Configuration
```python theme={null}
from laddr import Agent
from laddr.llms import gemini
from tools.web_tools import web_search
researcher = Agent(
name="researcher",
role="Web Research Specialist",
goal="Search the web and summarize findings",
backstory="""You are a research-focused agent who finds accurate
data online and presents it clearly.""",
llm=gemini(model="gemini-2.5-flash", temperature=0.0),
tools=[web_search],
max_retries=1,
max_iterations=3,
max_tool_calls=2,
timeout=45,
trace_enabled=True,
instructions="""
When researching:
1. Use web_search to find information
2. Prioritize recent sources
3. Cite sources in your response
"""
)
```
***
## Next Steps
* [Tool Schema](/reference/schemas/tool-schema) - Tool configuration schema
* [Agent Configuration](/guides/agents/agent-config) - Configuration guide
* [Agent Authoring](/guides/agents/authoring) - Best practices
# Tool Schema
Source: https://laddr.agnetlabs.com/reference/schemas/tool-schema
Complete schema reference for Laddr tool configuration - decorator parameters, schema generation, and examples
Complete reference for the Laddr tool configuration schema.
***
## Tool Definition
Laddr tools are Python functions decorated with `@tool`. Parameter schemas are **automatically generated** from function signatures, so you don't need to manually define JSON schemas.
### Basic Tool
```python theme={null}
from laddr import tool
@tool(
name="weather_lookup",
description="Fetches current weather for a given city"
)
def weather_lookup(city: str, units: str = "metric") -> dict:
"""Tool implementation."""
# Parameter schema is auto-generated from function signature
return {"status": "success", "data": result}
```
### With Pydantic Model
For complex validation, use a Pydantic BaseModel as the first parameter:
```python theme={null}
from laddr import tool
from pydantic import BaseModel
class WeatherInput(BaseModel):
city: str
units: str = "metric"
include_forecast: bool = False
@tool(
name="weather_lookup",
description="Fetches current weather for a given city"
)
def weather_lookup(inputs: WeatherInput) -> dict:
"""Tool implementation with Pydantic validation."""
# Inputs are automatically validated via Pydantic
return {"status": "success", "data": result}
```
### Minimal Tool (Auto-inferred)
If you omit `name` and `description`, they're inferred from the function:
```python theme={null}
@tool()
def web_search(query: str, max_results: int = 5) -> dict:
"""Search the web for information."""
# Name: "web_search" (from function name)
# Description: "Search the web for information." (from docstring)
return {"status": "success", "results": [...]}
```
***
## @tool Decorator Parameters
| Parameter | Type | Required | Description |
| ------------- | ----------- | -------- | --------------------------------------------------------- |
| `name` | `str` | ❌ | Unique tool identifier (auto-inferred from function name) |
| `description` | `str` | ❌ | What the tool does (auto-inferred from docstring) |
| `trace` | `bool` | ❌ | Enable tracing (default: `True`) |
| `trace_mask` | `list[str]` | ❌ | Fields to redact in traces |
**Note:** Parameter schemas are **automatically generated** from function signatures. You don't need to manually define JSON schemas.
### Parameter Descriptions
#### `name`
**Rules:**
* Must be unique and descriptive
* Use lowercase `snake_case`
* Avoid generic names
* If omitted, uses function name
**Examples:**
```python theme={null}
name="web_search"
name="calculate_area"
name="query_database"
```
#### `description`
**Best Practices:**
* Begin with an action verb
* Limit to 100 characters
* Avoid redundant phrasing like "use this tool to…"
* If omitted, uses first line of docstring
**Examples:**
```python theme={null}
description="Scrape text content from a webpage"
description="Translate text from English to French"
```
#### `trace`
Enables or disables trace logging.
**Example:**
```python theme={null}
trace=True # Default - all tool calls are traced
trace=False # Disable tracing for sensitive operations
```
Set to `False` when:
* Handling sensitive user data
* Reducing log volume for high-frequency tools
#### `trace_mask`
Redacts specified fields in traces.
**Example:**
```python theme={null}
trace_mask=["api_key", "token", "password"]
```
Fields matching these keys will appear as `***REDACTED***` in logs.
***
## Automatic Schema Generation
Parameter schemas are automatically generated from Python function signatures. The system maps Python types to JSON Schema types:
| Python Type | JSON Schema Type | Example |
| ------------- | -------------------- | ---------------------- |
| `str` | `string` | `query: str` |
| `int` | `integer` | `limit: int` |
| `float` | `number` | `price: float` |
| `bool` | `boolean` | `enabled: bool` |
| `list[T]` | `array` | `items: list[str]` |
| `dict` | `object` | `config: dict` |
| `Optional[T]` | Type with `null` | `timeout: int \| None` |
| `BaseModel` | `object` (validated) | `inputs: MyModel` |
### Type Hints Examples
```python theme={null}
# String parameter
def search(query: str) -> dict:
...
# Integer with default
def fetch(limit: int = 10) -> dict:
...
# Optional parameter
def process(timeout: int | None = None) -> dict:
...
# List parameter
def batch_process(items: list[str]) -> dict:
...
# Dict parameter
def configure(config: dict) -> dict:
...
# Complex nested (use Pydantic)
from pydantic import BaseModel
class Config(BaseModel):
host: str
port: int = 8080
ssl: bool = False
def setup(inputs: Config) -> dict:
...
```
### Supported Type Annotations
* **Basic types**: `str`, `int`, `float`, `bool`
* **Collections**: `list[T]`, `dict`, `tuple`
* **Optional**: `Optional[T]` or `T | None`
* **Union**: `str | int` (maps to first type)
* **Pydantic Models**: Full validation support
**Note:** Type annotations are optional but recommended for better schema generation.
***
## Complete Examples
### Example 1: Basic Tool
```python theme={null}
from laddr import tool
@tool(
name="weather_lookup",
description="Fetches current weather for a given city"
)
def weather_lookup(city: str, units: str = "metric") -> dict:
"""Get weather for a city."""
try:
result = get_weather(city, units)
return {"status": "success", "data": result}
except Exception as e:
return {"status": "error", "error": str(e)}
```
**Generated Schema:**
```json theme={null}
{
"type": "object",
"properties": {
"city": {"type": "string"},
"units": {"type": "string", "default": "metric"}
},
"required": ["city"]
}
```
### Example 2: With Pydantic Validation
```python theme={null}
from laddr import tool
from pydantic import BaseModel, Field
class WeatherInput(BaseModel):
city: str = Field(..., description="City name", min_length=1)
units: str = Field("metric", description="Temperature units", pattern="^(metric|imperial)$")
include_forecast: bool = Field(False, description="Include 5-day forecast")
@tool(
name="weather_lookup",
description="Fetches current weather for a given city"
)
def weather_lookup(inputs: WeatherInput) -> dict:
"""Get weather for a city with validation."""
try:
result = get_weather(inputs.city, inputs.units, inputs.include_forecast)
return {"status": "success", "data": result}
except Exception as e:
return {"status": "error", "error": str(e)}
```
### Example 3: API Wrapper Pattern
```python theme={null}
from laddr import tool
import requests
import os
@tool(
name="api_call",
description="Perform a GET request to external API",
trace_mask=["api_key"]
)
def api_call(endpoint: str, api_key: str | None = None) -> dict:
"""Call external API."""
key = api_key or os.getenv("API_KEY")
response = requests.get(
f"https://api.example.com/{endpoint}",
headers={"Authorization": f"Bearer {key}"},
timeout=10
)
return {"status": "success", "data": response.json()}
```
### Example 4: Data Transformer
```python theme={null}
from laddr import tool
import json
@tool(
name="convert_format",
description="Convert data between formats"
)
def convert_format(data: str, to_format: str = "json") -> dict:
"""Convert data between formats."""
if to_format == "json":
return {"status": "success", "result": json.loads(data)}
elif to_format == "csv":
return {"status": "success", "result": parse_csv(data)}
else:
return {"status": "error", "error": f"Unsupported format {to_format}"}
```
### Example 5: File Operations
```python theme={null}
from laddr import tool
import glob
@tool(
name="list_files",
description="List files by pattern"
)
def list_files(directory: str, pattern: str = "*") -> dict:
"""List files matching pattern."""
files = glob.glob(f"{directory}/{pattern}")
return {"status": "success", "files": files, "count": len(files)}
```
***
## Return Format
Tools should return dictionaries with a consistent format:
```python theme={null}
# Success
return {
"status": "success",
"data": {...}
}
# Error
return {
"status": "error",
"error": "Error message"
}
```
**Status Values:**
* `"success"` - Tool executed successfully
* `"error"` - Tool execution failed
## Registering Tools with Agents
Tools must be registered with agents. There are several ways to do this:
### Method 1: Direct Registration
```python theme={null}
from laddr import Agent
from tools.web_tools import web_search
from tools.math_tools import calculate
agent = Agent(
name="researcher",
tools=[web_search, calculate]
)
```
### Method 2: Auto-Discovery
Tools are automatically discovered from `agents..tools` package:
```
agents/
researcher/
tools/
__init__.py
web_search.py
calculate.py
```
The agent automatically loads all `@tool` decorated functions from this package.
### Method 3: Using bind\_tools
```python theme={null}
from laddr.core.tooling import bind_tools
from agents.researcher.tools import web_search
class ResearcherAgent(Agent):
def __init__(self):
super().__init__(name="researcher")
bind_tools(self, [web_search, "summarize"]) # String names auto-import
```
## Tool Registry
Tools are managed through a `ToolRegistry` that supports:
* **Registration**: Add tools with optional aliases
* **Lookup**: Find tools by name or alias
* **Listing**: Get all available tools
* **Validation**: Automatic input validation via Pydantic
```python theme={null}
from laddr.core.tooling import ToolRegistry
registry = ToolRegistry()
registry.register(web_search, aliases=["search", "web"])
tool = registry.get("web_search") # or "search" or "web"
```
## Testing Tools
### Manual Testing (Python)
```python theme={null}
from tools.web_tools import web_search
result = web_search(query="AI trends", max_results=3)
print(result)
```
### Testing via API
```bash theme={null}
curl -X GET "http://localhost:8000/api/agents/researcher/tools"
```
Returns all tools with their schemas:
```json theme={null}
{
"agent": "researcher",
"tools": [
{
"name": "web_search",
"description": "Search the web for information",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string"},
"max_results": {"type": "integer", "default": 5}
},
"required": ["query"]
}
}
]
}
```
***
## Best Practices
1. **Use Type Hints**: Always annotate function parameters for better schema generation
2. **Descriptive Names**: Use clear, action-oriented tool names (`web_search` not `search`)
3. **Error Handling**: Always return consistent error format
4. **Sensitive Data**: Use `trace_mask` for API keys, tokens, passwords
5. **Pydantic for Complex**: Use Pydantic models for complex validation needs
6. **Documentation**: Write clear docstrings (used as descriptions if not provided)
## Common Patterns
### Pattern: API Wrapper
```python theme={null}
@tool(name="api_call", description="Call external API")
def api_call(endpoint: str, method: str = "GET") -> dict:
# Wraps external API calls
```
### Pattern: Data Transformer
```python theme={null}
@tool(name="transform", description="Transform data format")
def transform(data: str, format: str) -> dict:
# Converts between data formats
```
### Pattern: File Operations
```python theme={null}
@tool(name="file_op", description="Perform file operation")
def file_op(path: str, operation: str) -> dict:
# File system operations
```
## Next Steps
* [Tool Configuration](/tool-config) - Complete tool development guide
* [Agent Configuration](/agent-config) - Agent configuration schema
* [MCP Integration](/guides/mcp-integration) - Connect to MCP servers
* [API Reference](/api-reference) - API endpoints for tool inspection