Universal Tool-Calling
for Non-Tool-Native LLMs

Why toolproxy?

Installation

# Core (no optional deps)
pip install toolproxy

# With FastAPI integration
pip install "toolproxy[fastapi]"

# With Anthropic adapter
pip install "toolproxy[anthropic]"

# Everything
pip install "toolproxy[full]"

Requires Python 3.10+

Quick Start

from toolproxy import UniversalAgent, tool

@tool
def search_web(query: str) -> str:
    """Search the web for recent information."""
    return f"Top result for '{query}': ..."

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    return str(eval(expression))

agent = UniversalAgent(
    model="openrouter/mistralai/mistral-7b-instruct",
    tools=[search_web, calculate],
)

result = agent.run("What is 15% of 240, and who invented Python?")
print(result.content)
print(f"Steps taken: {result.steps_taken}")

How It Works

UniversalAgent.run(prompt)
    │
    ├── Planner  ──  auto-detects native vs emulated mode
    │       ├── Native   → provider tool-call API (OpenAI format)
    │       └── Emulated → structured JSON schema injected into system prompt
    │
    ├── Executor  ──  validates args, runs tool, captures errors
    │       └── (optional) ToolCache  →  skip if cached result exists
    │
    └── LoopController  ──  repeats until "final" answer or max_steps
            └── (optional) BaseTracer  →  observability hooks

Emulated mode protocol — when native tools aren't available, the model is prompted to output exactly one of:

{"type": "tool_call", "tool": {"tool_name": "get_weather", "arguments": {"city": "Chennai"}}}
{"type": "final", "content": "The weather in Chennai is Sunny, 25°C."}

Malformed responses are automatically retried (up to parse_retries times, default 3) with a corrective error message.

Supported Models & Adapters

Core API

The @tool Decorator

Mark any Python function as a tool. Type hints generate the argument schema automatically:

from toolproxy import tool

@tool
def get_stock_price(ticker: str, currency: str = "USD") -> str:
    """Get the current stock price for a ticker symbol."""
    return f"{ticker}: $182.50 {currency}"

# Async tools work too
@tool
async def fetch_data(url: str) -> str:
    """Fetch data from a URL asynchronously."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
        return resp.text

For explicit schemas:

from pydantic import BaseModel
from toolproxy import tool

class SearchArgs(BaseModel):
    query: str
    max_results: int = 5

@tool(name="web_search", description="Search the web", args_schema=SearchArgs)
def web_search(query: str, max_results: int = 5) -> str:
    ...

UniversalAgent

from toolproxy import UniversalAgent
from toolproxy.cache import InMemoryToolCache
from toolproxy.memory import ConversationBuffer
from toolproxy.observability import PrintTracer

agent = UniversalAgent(
    model="openrouter/mistralai/mistral-7b-instruct",
    tools=[get_weather, search_web],

    # Mode: "auto" (default), "native_only", "emulated_only"
    mode="auto",

    # Max agent loop iterations
    max_steps=10,

    # API key (or set OPENROUTER_API_KEY env var)
    api_key="sk-...",

    # Custom base URL for OpenAI-compatible endpoints
    base_url="https://my-proxy.example.com/v1",

    # Multi-turn memory (v0.3+)
    memory=ConversationBuffer(),

    # Observability (v0.3+)
    tracer=PrintTracer(),

    # Tool result cache (v0.4+)
    cache=InMemoryToolCache(ttl=300),
)

run() and arun()

# Synchronous
result = agent.run("What is the weather in Chennai?")

# Asynchronous
result = await agent.arun("What is the weather in Chennai?")

# With options
result = agent.run(
    "Tell me about today's AI news",
    return_trace=True,           # include full tool-call trace
    max_steps=5,                 # override for this run only (v0.4)
    extra_tools=[my_new_tool],   # ad-hoc tools for this run only (v0.4)
    on_tool_call=lambda step, tc: print(f"→ Calling {tc.tool_name}"),
    on_tool_result=lambda step, tr: print(f"← Result: {tr.output}"),
    on_model_output=lambda step, text: print(f"Model: {text}"),
)

print(result.content)       # final answer
print(result.steps_taken)   # number of loop iterations
print(result.trace)         # AgentTrace (if return_trace=True)

Streaming

# Sync streaming
for chunk in agent.stream("Tell me about the solar system"):
    if chunk.tool_call:
        print(f"Calling: {chunk.tool_call.tool_name}")
    elif chunk.tool_result:
        print(f"Result: {chunk.tool_result.output}")
    elif chunk.delta:
        print(chunk.delta, end="", flush=True)
    elif chunk.done:
        print()  # final newline

# Async streaming
async for chunk in agent.astream("Tell me about the solar system"):
    ...

v0.4 Features

Tool Result Caching

Avoid redundant tool calls when the same tool is invoked with identical arguments:

from toolproxy.cache import InMemoryToolCache, FileToolCache

# In-memory: fast, TTL-based, lost on restart
agent = UniversalAgent(
    model="...",
    tools=[search_web],
    cache=InMemoryToolCache(
        ttl=300,        # seconds (None = no expiry)
        max_size=512,   # max entries (LRU eviction)
    ),
)

# File-backed: survives process restarts
agent = UniversalAgent(
    model="...",
    tools=[search_web],
    cache=FileToolCache("tool_cache.json", ttl=3600),
)

# Check cache stats
stats = agent.cache.stats()
# → {"hits": 12, "misses": 3, "size": 8}

# Invalidate a specific entry
agent.cache.invalidate("search_web", {"query": "old topic"})

# Clear all cached results
agent.cache.clear()

How caching works: The (tool_name, arguments) pair is hashed with SHA-256. On a hit, the tool is skipped entirely and the cached string is returned as the tool result. Only successful results are cached — errors are never stored.

Retry with Backoff

Automatically recover from transient failures (network drops, HTTP 429/500/503):

from toolproxy.retry import RetryConfig, with_retry, awith_retry, HTTP_RETRY_CONFIG

# Custom config
config = RetryConfig(
    max_attempts=4,
    initial_delay=1.0,    # seconds before first retry
    backoff_factor=2.0,   # doubles on each retry: 1s, 2s, 4s
    max_delay=30.0,       # capped at 30 seconds
    jitter=True,          # adds ≤10% random jitter
    retryable_exceptions=(ConnectionError, TimeoutError),
)

# Wrap any callable
result = with_retry(lambda: expensive_api_call(), config=config)

# Async version
result = await awith_retry(lambda: async_api_call(), config=config)

# Pre-built config for HTTP model calls
from toolproxy.retry import HTTP_RETRY_CONFIG
result = with_retry(lambda: client.generate(messages), config=HTTP_RETRY_CONFIG)

Persistent File Memory

Keep conversation history across process restarts:

from toolproxy.memory import JsonFileMemory

agent = UniversalAgent(
    model="openrouter/mistralai/mistral-7b-instruct",
    tools=[...],
    memory=JsonFileMemory(
        "chat_history.json",
        max_messages=50,        # keep last 50 non-system messages
        preserve_system=True,   # system messages are never trimmed
    ),
)

# Turn 1 (run 1)
agent.run("My name is Alice.")

# Turn 2 (run 2, after restart) — Alice is still remembered
agent.run("What's my name?")
# → "Your name is Alice."

Writes are atomic (.tmp → rename), so history is never corrupted by a crash.

Agent Extensions

Dynamically modify agents without rebuilding them:

# Add a tool after construction
@tool
def translate(text: str, language: str) -> str:
    """Translate text to a language."""
    ...

agent.add_tool(translate)

# Full reset: clear memory + cache
agent.reset()

# Per-run max_steps override (restores after run)
result = agent.run("complex task", max_steps=25)

# Per-run extra tools (removed after run, even on error)
result = agent.run(
    "use the special tool",
    extra_tools=[one_off_tool],
)

# Check if a tool is registered
"translate" in agent.registry  # → True
len(agent.registry)             # → 3

# Access cache stats
agent.cache.stats()

FastAPI Integration

Expose your agent as a production-ready HTTP service:

from fastapi import FastAPI
from toolproxy import UniversalAgent, tool
from toolproxy.integrations.fastapi import create_agent_router

@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Results for {query}..."

agent = UniversalAgent(
    model="openrouter/mistralai/mistral-7b-instruct",
    tools=[search],
)

app = FastAPI(title="My Agent API")
app.include_router(
    create_agent_router(
        agent,
        api_key="my-secret-key",  # optional X-API-Key header auth
    ),
    prefix="/agent",
)

Endpoints created:

SSE stream events:

{"type": "tool_call",   "tool": "search", "args": {"query": "AI"}}
{"type": "tool_result", "tool": "search", "output": "AI is ..."}
{"type": "delta",       "text": "Based on "}
{"type": "delta",       "text": "the search..."}
{"type": "done",        "content": "Based on the search, AI is ..."}

Test with curl:

curl -X POST http://localhost:8000/agent/run \
     -H "Content-Type: application/json" \
     -H "X-API-Key: my-secret-key" \
     -d '{"prompt": "What is the weather in Chennai?"}'

v0.5 Features

Parallel Emulated Tools

For models without native function calling support, toolproxy emulates tool calling by guiding the model with an advanced structured prompt. Previously, emulated mode was limited to executing one tool at a time. In v0.5, emulated mode can request and execute multiple tools in parallel within a single step:

agent = UniversalAgent(
    model="openrouter/google/gemini-2.5-flash",  # emulated tool mode
    tools=[get_weather, get_stock_price],
    allow_parallel_emulated=True,  # enabled by default
)

# If both tools are requested, they execute concurrently in parallel
await agent.arun("What is the weather in Chennai and the stock price of GOOG?")

Under the hood, the planner requests a list of JSON action structures in a single prompt response, and the executor dispatches them concurrently.

SQLite Memory Backend

SqliteMemory is a persistent memory provider backed by a local SQLite database file. It enables multi-session isolation, keyword searching, and history trimming.

from toolproxy.memory import SqliteMemory

memory = SqliteMemory(
    db_path="chat_history.db",
    session_id="session-user-42",
    max_messages=20,       # trim older messages when limit is reached
    preserve_system=True,  # keep the system prompt intact when trimming
)

agent = UniversalAgent(
    model="...",
    memory=memory,
)

Additional Features:

• Search: Query the database for past messages using SQL LIKE wildcard matching.

# Search within the current session:
results = memory.search("Python")

# Search across all sessions by passing session_id="":
all_results = memory.search("Python", session_id="")

# Get all active session IDs stored in the database:
active_sessions = memory.sessions()

# Query the number of messages:
count = memory.message_count()  # or len(memory)

Structured Output Agent

StructuredOutputAgent wraps a UniversalAgent and ensures that the final response validates against a specific Pydantic model instead of a raw text string. It automatically appends the JSON schema requirements to the prompt, validates the model output, and performs an automatic format-retry loop if validation fails.

from pydantic import BaseModel
from toolproxy import UniversalAgent
from toolproxy.structured import StructuredOutputAgent

class MovieDetails(BaseModel):
    title: str
    release_year: int
    genres: list[str]
    rating: float

agent = UniversalAgent(model="openai/gpt-4o")
structured_agent = StructuredOutputAgent(agent, output_schema=MovieDetails, max_retries=1)

response = structured_agent.run("Tell me about Inception")
# response is a StructuredAgentResponse containing the validated model instance
movie = response.data
print(f"{movie.title} ({movie.release_year}) - Rating: {movie.rating}")

Tool Timeout & Concurrency

Ensure slow external tools don't freeze your agent, and limit concurrency to prevent rate-limiting or overloading external APIs.

agent = UniversalAgent(
    model="...",
    tools=[slow_tool, other_tool],
    tool_timeout=5.0,     # cancel any tool execution exceeding 5.0 seconds
    max_concurrency=3,    # run at most 3 tools in parallel concurrently
)

When a tool exceeds the timeout, toolproxy raises a ToolTimeoutError and logs the event, reporting the failure back to the agent so it can adapt or retry.

Interactive Chat REPL

Quickly interact with your agent in a command-line interface. It handles multi-turn conversation memory, streams response tokens as they arrive, and prints visual indicators when tools are called.

agent = UniversalAgent(model="openai/gpt-4o", tools=[get_weather])

# Start the REPL
agent.chat(
    welcome_msg="Welcome to the Interactive Agent Chat! Type 'quit' to exit.",
    stream=True,  # stream tokens as they arrive
)

LangChain Tool Bridge

The LangChain bridge allows you to reuse LangChain tools within toolproxy and vice-versa, allowing for easy migration and interoperability.

from toolproxy.integrations.langchain import from_langchain_tool, to_langchain_tool

# 1. Use a LangChain tool in toolproxy:
from langchain_community.tools import WikipediaQueryRun
wikipedia = WikipediaQueryRun(...)

tp_tool = from_langchain_tool(wikipedia)
agent = UniversalAgent(model="...", tools=[tp_tool])

# 2. Use a toolproxy tool in LangChain:
from toolproxy import tool

@tool
def custom_calc(a: int, b: int) -> int:
    "Add two numbers."
    return a + b

lc_tool = to_langchain_tool(custom_calc)
# Now pass lc_tool to your LangChain agent's tools list!

Memory

Keep context across multiple agent.run() calls:

from toolproxy.memory import (
    ConversationBuffer,    # unlimited history
    SlidingWindowMemory,   # last N messages
    TokenWindowMemory,     # fits within token budget
    JsonFileMemory,        # persistent across restarts (v0.4)
)

# Unlimited
memory = ConversationBuffer()

# Last 20 messages (keeps system messages)
memory = SlidingWindowMemory(max_messages=20)

# Within 4096 tokens (uses char/4 approximation)
memory = TokenWindowMemory(max_tokens=4096)

# Persistent JSON file
memory = JsonFileMemory("history.json", max_messages=50)

agent = UniversalAgent(model="...", tools=[...], memory=memory)

agent.run("My name is Alice.")
agent.run("What's my name?")   # remembers: "Your name is Alice."

agent.clear_memory()           # wipe history without rebuilding agent

Observability & Tracing

from toolproxy.observability import FileTracer, PrintTracer, OpenTelemetryTracer

# Print to console
agent = UniversalAgent(..., tracer=PrintTracer(prefix="[agent]"))

# Write structured JSON to a file
agent = UniversalAgent(..., tracer=FileTracer("runs.jsonl", mode="append"))

# Emit OpenTelemetry spans
from opentelemetry import trace
tracer_provider = trace.get_tracer_provider()
agent = UniversalAgent(..., tracer=OpenTelemetryTracer(tracer_provider))

# Read recorded runs from a FileTracer
tracer = FileTracer("runs.jsonl", mode="append")
runs = tracer.read_runs()
for run in runs:
    print(run["prompt"], run["steps"])

Execution Policies

Control which tools the model is allowed to call:

from toolproxy.config import ExecutionPolicy

# Allow all tools (default)
policy = ExecutionPolicy(mode="allow_all")

# Whitelist specific tools
policy = ExecutionPolicy(mode="allow_only", allowed_tools=["get_weather"])

# Require confirmation before dangerous tools
policy = ExecutionPolicy(
    mode="confirm_before",
    confirm_tools=["delete_file", "send_email"],
)

agent = UniversalAgent(
    model="...",
    tools=[...],
    execution_policy=policy,
    confirm_callback=lambda tool_name, args: input(f"Allow {tool_name}({args})? [y/N] ") == "y",
)

MCP Support

Connect to any Model Context Protocol server and use its tools:

import asyncio
from toolproxy import UniversalAgent
from toolproxy.mcp import MCPToolset

async def main():
    # Discover tools from an MCP server
    toolset = await MCPToolset.from_server("http://localhost:3000")
    print(f"Discovered {len(toolset)} tools: {[t.name for t in toolset.tools]}")

    agent = UniversalAgent(
        model="openrouter/mistralai/mistral-7b-instruct",
    )
    toolset.register_into(agent.registry)

    result = await agent.arun("Use the MCP tools to answer my question.")
    print(result.content)

asyncio.run(main())

Async & Parallel Tools

toolproxy fully supports async — including parallel tool dispatch:

import asyncio
from toolproxy import UniversalAgent, tool

@tool
async def fetch_page(url: str) -> str:
    """Fetch a webpage asynchronously."""
    async with httpx.AsyncClient() as c:
        return (await c.get(url)).text

agent = UniversalAgent(
    model="openrouter/mistralai/mistral-7b-instruct",
    tools=[fetch_page],
)

# When the model requests multiple tools at once (native mode),
# they are dispatched concurrently via asyncio.gather
result = await agent.arun("Fetch the homepages of openai.com and anthropic.com")

Environment Variables

Changelog