Key Features

# Scale workers
laddr scale researcher 5
laddr scale coordinator 3

# .env
QUEUE_BACKEND=redis  # or kafka, memory

{
  "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": "..."}
  ]
}

# View logs
laddr logs researcher --follow
laddr logs coordinator --tail 100

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]
)

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]
)

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(...)

# .env
STORAGE_BACKEND=minio
MINIO_ENDPOINT=localhost:9000
MINIO_ACCESS_KEY=minioadmin
MINIO_SECRET_KEY=minioadmin

# .env
DB_BACKEND=postgresql
DATABASE_URL=postgresql://user:pass@localhost:5432/laddr

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"))

POST /api/prompts
Content-Type: application/json

{
  "prompt_name": "researcher",
  "inputs": {"query": "Latest AI trends"}
}

GET /api/prompts/{prompt_id}

GET /api/health

# In coordinator agent
result = await self.delegate_task(
    agent_name="researcher",
    task_description="Research AI trends",
    task="research",
    task_data={"topic": "AI trends"}
)

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}}
    ]
)

agent = Agent(
    name="researcher",
    max_retries=3,  # Retry failed tasks
    timeout=60      # Timeout in seconds
)

{
  "event_type": "tool_error",
  "tool": "web_search",
  "error": "API rate limit exceeded",
  "retry_count": 1
}