LLM Output
        ↓
CHUK Tool Processor
        ↓
 ┌──────────────┬────────────────────┐
 │ Local Tools  │ Remote Tools (MCP) │
 └──────────────┴────────────────────┘

    LLM Output
        ↓
Parsers (XML / OpenAI / JSON)
        ↓
┌─────────────────────────────┐
│   Execution Middleware      │
│  (Applied in this order)    │
│   • Cache                   │
│   • Rate Limit              │
│   • Retry (with backoff)    │
│   • Circuit Breaker         │
│   • Bulkhead                │
└─────────────────────────────┘
        ↓
   Execution Strategy
   ┌──────────────────────┐
   │ • InProcess          │  ← Fast, trusted
   │ • Isolated/Subprocess│  ← Safe, untrusted
   │ • Remote via MCP     │  ← Distributed
   └──────────────────────┘

pip install chuk-tool-processor

# Or with uv (recommended)
uv pip install chuk-tool-processor

import asyncio
from chuk_tool_processor import ToolProcessor, create_registry

class Calculator:
    async def execute(self, operation: str, a: float, b: float) -> dict:
        ops = {"add": a + b, "multiply": a * b, "subtract": a - b}
        return {"result": ops.get(operation, 0)}

async def main():
    registry = create_registry()
    await registry.register_tool(Calculator, name="math.calculator")  # Dotted name → namespace="math"

    async with ToolProcessor(registry=registry, enable_caching=True, enable_retries=True) as p:
        # Works with OpenAI, Anthropic, or JSON formats
        result = await p.process('<tool name="math.calculator" args=\'{"operation": "multiply", "a": 15, "b": 23}\'/>')
        print(result[0].result)  # {'result': 345}

asyncio.run(main())

# These are equivalent:
await registry.register_tool(FetchUser, name="web.fetch_user")           # Auto-parsed
await registry.register_tool(FetchUser, name="fetch_user", namespace="web")  # Explicit

# Call using the full dotted name
result = await processor.process([{"tool": "web.fetch_user", "arguments": {"user_id": "123"}}])

# Anthropic XML format
anthropic_output = '<tool name="search" args=\'{"query": "Python"}\'/>'

# OpenAI tool_calls format
openai_output = {
    "tool_calls": [{
        "type": "function",
        "function": {"name": "search", "arguments": '{"query": "Python"}'}
    }]
}

# Direct JSON
json_output = [{"tool": "search", "arguments": {"query": "Python"}}]

# All work identically
results = await processor.process(anthropic_output)
results = await processor.process(openai_output)
results = await processor.process(json_output)

async with ToolProcessor(
    # Execution settings
    default_timeout=30.0,
    max_concurrency=20,

    # Reliability features
    enable_caching=True,
    cache_ttl=600,
    enable_rate_limiting=True,
    global_rate_limit=100,
    tool_rate_limits={"expensive_api": (5, 60)},  # 5 req/min
    enable_retries=True,
    max_retries=3,
    enable_circuit_breaker=True,
    circuit_breaker_threshold=5,

    # Multi-tenant isolation
    enable_bulkhead=True,
    bulkhead_config=BulkheadConfig(
        default_limit=10,
        tool_limits={"slow_api": 2},
        patterns={"db.*": 3, "mcp.notion.*": 2},  # Pattern-based limits
    ),
) as processor:
    # Execute with request context
    ctx = ExecutionContext(
        request_id="req-123",
        user_id="user-456",
        tenant_id="acme-corp",
    )
    results = await processor.process(llm_output, context=ctx)

from chuk_tool_processor import ToolProcessor, ReturnOrder

async with ToolProcessor() as processor:
    # Results return as tools complete (fast tools first) - default
    results = await processor.process(calls, return_order="completion")

    # Results return in submission order (deterministic)
    results = await processor.process(calls, return_order="submission")

from chuk_tool_processor import (
    GreedyDagScheduler,
    SchedulingConstraints,
    ToolCallSpec,
    ToolMetadata,
)

scheduler = GreedyDagScheduler()

# Define calls with dependencies
calls = [
    ToolCallSpec(call_id="fetch", tool_name="api.fetch",
                 metadata=ToolMetadata(pool="web", est_ms=300)),
    ToolCallSpec(call_id="transform", tool_name="compute.transform",
                 depends_on=("fetch",)),
    ToolCallSpec(call_id="store", tool_name="db.write",
                 depends_on=("transform",)),
]

# Plan execution with constraints
constraints = SchedulingConstraints(
    deadline_ms=5000,
    pool_limits={"web": 2, "db": 1},
)
plan = scheduler.plan(calls, constraints)

# plan.stages: (('fetch',), ('transform',), ('store',))
# plan.skip: () or low-priority calls that would miss deadline

from chuk_tool_processor.mcp import setup_mcp_http_streamable

# Cloud services (Notion, etc.)
processor, manager = await setup_mcp_http_streamable(
    servers=[{
        "name": "notion",
        "url": "https://mcp.notion.com/mcp",
        "headers": {"Authorization": f"Bearer {token}"}
    }],
    namespace="notion",
    enable_caching=True,
    enable_retries=True
)

# Use remote tools
results = await processor.process(
    '<tool name="notion.search_pages" args=\'{"query": "docs"}\'/>'
)

from chuk_tool_processor.mcp.middleware import (
    MiddlewareConfig,
    MiddlewareStack,
    RetrySettings,
    CircuitBreakerSettings,
    RateLimitSettings,
)

# Configure middleware layers
config = MiddlewareConfig(
    retry=RetrySettings(max_retries=3, base_delay=1.0),
    circuit_breaker=CircuitBreakerSettings(failure_threshold=5),
    rate_limiting=RateLimitSettings(enabled=True, global_limit=100),
)

# Wrap StreamManager with middleware
middleware = MiddlewareStack(stream_manager, config=config)

# Execute with automatic retry, circuit breaking, and rate limiting
result = await middleware.call_tool("notion.search", {"query": "docs"})

# Enable Redis for everything
export CHUK_REGISTRY_BACKEND=redis
export CHUK_RESILIENCE_BACKEND=redis
export CHUK_REDIS_URL=redis://localhost:6379/0

# Enable resilience features
export CHUK_CIRCUIT_BREAKER_ENABLED=true
export CHUK_RATE_LIMIT_ENABLED=true
export CHUK_RATE_LIMIT_GLOBAL=100

from chuk_tool_processor import ProcessorConfig

# Load from environment and create fully-configured processor
config = ProcessorConfig.from_env()
processor = await config.create_processor()

async with processor:
    results = await processor.process(llm_output)

from chuk_tool_processor import ProcessorConfig, RegistryConfig, BackendType
from chuk_tool_processor.config import CircuitBreakerConfig, RateLimitConfig

config = ProcessorConfig(
    # Registry and resilience use Redis
    registry=RegistryConfig(backend=BackendType.REDIS),
    resilience_backend=BackendType.REDIS,
    redis_url="redis://localhost:6379/0",

    # Enable features
    circuit_breaker=CircuitBreakerConfig(enabled=True, failure_threshold=5),
    rate_limit=RateLimitConfig(enabled=True, global_limit=100),
)

processor = await config.create_processor()

pip install chuk-tool-processor[redis]  # or: uv add chuk-tool-processor[redis]

from chuk_tool_processor.guards import (
    GuardChain,
    SchemaStrictnessGuard,
    SensitiveDataGuard,
    NetworkPolicyGuard,
    ConcurrencyGuard,
)

# Create individual guards
schema_guard = SchemaStrictnessGuard(get_schema=my_schema_getter)
sensitive_guard = SensitiveDataGuard()  # Detects API keys, JWTs, etc.
network_guard = NetworkPolicyGuard(block_private_ips=True)
concurrency_guard = ConcurrencyGuard(global_max=50, per_tool_max={"heavy_api": 2})

# Compose into a chain
chain = GuardChain([schema_guard, sensitive_guard, network_guard, concurrency_guard])

# Check before execution
result = await chain.check_all_async("api.fetch", {"url": "https://example.com"})
if result.blocked:
    print(f"Blocked by {result.stopped_at}: {result.reason}")

from chuk_tool_processor.discovery import ToolSearchEngine, BaseDynamicToolProvider

# Create a search engine for your tools
engine = ToolSearchEngine()
engine.set_tools(my_tools)

# Natural language search with synonym expansion
results = engine.search("gaussian distribution")  # Finds "normal_cdf"
results = engine.search("find the average")       # Finds "calculate_mean"
results = engine.search("multipley")              # Finds "multiply" (typo tolerance)

# Session boosting - recently used tools rank higher
engine.record_tool_use("calculate_mean", success=True)
engine.advance_turn()
results = engine.search("calculate")  # "calculate_mean" now boosted

class MyToolProvider(BaseDynamicToolProvider):
    async def get_all_tools(self) -> list[Tool]:
        return self._tools

    async def execute_tool(self, name: str, args: dict) -> dict:
        return await self._tools[name].execute(**args)

provider = MyToolProvider()

# LLM gets 4 meta-tools: list_tools, search_tools, get_tool_schema, call_tool
tools_for_llm = provider.get_dynamic_tools()

# LLM workflow: search → get schema → call
results = await provider.search_tools("calculate average")
schema = await provider.get_tool_schema("calculate_mean")
result = await provider.call_tool("calculate_mean", {"values": [1, 2, 3]})

from chuk_tool_processor.observability import setup_observability

setup_observability(
    service_name="my-tool-service",
    enable_tracing=True,     # → OpenTelemetry traces
    enable_metrics=True,     # → Prometheus metrics at :9090/metrics
    metrics_port=9090
)
# Every tool execution is now automatically traced and metered

from chuk_tool_processor.core.exceptions import ErrorCategory

results = await processor.process(llm_output)
for result in results:
    if result.error_info:
        match result.error_info.category:
            case ErrorCategory.RATE_LIMIT:
                await asyncio.sleep(result.retry_after_ms / 1000)
                return await retry()
            case ErrorCategory.CIRCUIT_OPEN:
                return await use_fallback_tool()
            case _ if not result.retryable:
                return await report_permanent_failure()

# Getting started
python examples/01_getting_started/hello_tool.py

# Dynamic tool discovery (search, synonyms, fuzzy matching)
python examples/07_discovery/dynamic_tools_demo.py

# Hero demo: 8 tools, 5-second deadline, 3 pools (DAG + bulkheads + context)
python examples/02_production_features/hero_runtime_demo.py

# Production patterns (bulkheads, context, scoped registries)
python examples/02_production_features/production_patterns_demo.py

# Runtime features (return order, pattern bulkheads, scheduling)
python examples/02_production_features/runtime_features_demo.py

# Structured error handling for planners
python examples/02_production_features/structured_errors_demo.py

# Runtime guards (validation, security, resource limits)
python examples/guards_demo.py

# Redis registry for distributed deployments
python examples/02_production_features/redis_registry_demo.py

# Distributed configuration (Redis registry + resilience)
python examples/02_production_features/distributed_config_demo.py

# Observability demo
python examples/02_production_features/observability_demo.py

# MCP integration
python examples/04_mcp_integration/stdio_echo.py
python examples/04_mcp_integration/notion_oauth.py
python examples/04_mcp_integration/middleware_demo.py

# Core package
pip install chuk-tool-processor

# With observability (OpenTelemetry + Prometheus)
pip install chuk-tool-processor[observability]

# With MCP support
pip install chuk-tool-processor[mcp]

# With Redis registry (distributed deployments)
pip install chuk-tool-processor[redis]

# With fast JSON (2-3x faster with orjson)
pip install chuk-tool-processor[fast-json]

# All extras
pip install chuk-tool-processor[all]

┌─────────────────────────────────────────────────────┐
│  Your Agent / LangChain / LlamaIndex / Custom       │  ← Decides WHICH tools
└─────────────────────────────────────────────────────┘
                          ↓
┌─────────────────────────────────────────────────────┐
│            CHUK Tool Processor                      │  ← Executes tools RELIABLY
│  (timeouts, retries, caching, rate limits, etc.)   │
└─────────────────────────────────────────────────────┘
                          ↓
┌─────────────────────────────────────────────────────┐
│          Local Tools / MCP Servers                  │  ← Does the actual work
└─────────────────────────────────────────────────────┘

# Development setup
git clone <repository-url>
cd chuk-tool-processor
uv pip install -e ".[dev]"

# Run tests
make check