toolops 1.0.0

The Industrial-Grade Resilience & Efficiency Layer for AI Agent Tools

🇬🇧 English | 🇫🇷 Français | 🇨🇳 中文 | 🇯🇵 日本語 | 🇪🇸 Español | 🇩🇪 Deutsch | 🇵🇹 Português | 🇰🇷 한국어 | 🇷🇺 Русский | 🇮🇳 हिन्दी

ToolOps

The Industrial-Grade Resilience & Efficiency Layer for AI Agent Tools

Build production-ready AI agents. Stop writing infrastructure boilerplate.

Website · Documentation · Quickstart · Changelog

---

⚡ 30-Second Pitch

"ToolOps is to AI Tools what a Service Mesh is to Microservices."

When you build AI agents, external calls (LLMs, APIs, DBs) are expensive, unreliable, and slow. ToolOps eliminates the boilerplate. It is a framework-agnostic middleware SDK that wraps any Python function in a single decorator, instantly upgrading it with caching, resilience, observability, and concurrency control.

# BEFORE ToolOps: 80+ lines of cache managers, retry logic, circuit breakers...

# AFTER ToolOps:
@readonly(cache_backend="semantic", cache_ttl=3600, retry_count=3)
async def ask_llm(query: str) -> str:
    return await llm.complete(query)  # Automatically cached, retried, and traced

🚀 Benchmarks & Impact

• 90% reduction in LLM calls via Semantic Caching.
• <5ms overhead per tool call execution.
• 0 code changes to your core business logic.

---

⚖️ Why ToolOps?

Every agent developer hits a wall when moving from demo to production. Here is how ToolOps compares to standard alternatives:

Feature	Standard @lru_cache	Framework-Native	🚀 ToolOps v1.0.0
Async / await support	❌	✅	✅ Native
Semantic (meaning-aware) cache	❌	⚠️ Basic	✅ Advanced Embeddings
Distributed / Persistent cache	❌	⚠️ Varies	✅ Postgres, SQLite, MySQL, Valkey/Redis
Circuit Breaker	❌	❌	✅ Native
Automatic Retries w/ Backoff	❌	⚠️ Plugin required	✅ Native
Request Coalescing (Anti-Thundering Herd)	❌	❌	✅ Native
Stale-if-error Fallback	❌	❌	✅ Native
Security (SHA-256 keys, Auto-masking)	❌	❌	✅ Native
OpenTelemetry & Prometheus	❌	⚠️ Callbacks needed	✅ Native
Framework Agnostic	✅	❌ Locked-in	✅ 100% Universal

---

📦 Installation

ToolOps is fully batteries-included. Installing it installs all cache backends (Memory, File, SQLite, Valkey, Redis, MySQL/MariaDB, Postgres, and Semantic), resilience features, and OpenTelemetry/Prometheus observability tooling by default.

pip install toolops

💻 OS-Specific Guides

We strongly recommend isolating your project in a virtual environment.

🐧 Linux & 🍎 macOS

# 1. Create and activate a virtual environment
python -m venv .venv
source .venv/bin/activate

# 2. Install ToolOps
pip install toolops

# 3. Verify installation
toolops doctor

🪟 Windows (PowerShell)

# 1. Create and activate a virtual environment
python -m venv .venv
.venv\Scripts\Activate.ps1

# 2. Install ToolOps
pip install toolops

# 3. Verify installation
toolops doctor

🪟 Windows (Command Prompt)

:: 1. Create and activate a virtual environment
python -m venv .venv
.venv\Scripts\activate.bat

:: 2. Install ToolOps
pip install toolops

:: 3. Verify installation
toolops doctor

---

🚀 Quickstart

This minimal example gets you from install to a working, cached, resilient tool in under 2 minutes.

# Imports
import asyncio

from toolops.cache import MemoryCache
from toolops import readonly, sideeffect, cache_manager


# Step 1: Register a cache backend (do this once at startup)
cache_manager.register("memory", MemoryCache(), is_default=True)


# Step 2: Decorate any async function with @readonly for read operations
@readonly(cache_backend="memory", cache_ttl=3600, retry_count=3)
async def fetch_weather(city: str) -> dict:
    # Simulate an external API call
    return {"city": city, "temp": 22, "condition": "sunny"}


# Step 3: Decorate write operations with @sideeffect (no caching, but protected)
@sideeffect(circuit_breaker=True, timeout=5.0, retry_count=2)
async def send_alert(message: str) -> bool:
    # Simulate sending a notification
    print(f"Alert sent: {message}")
    return True


async def main():
    # First call hits the API
    result = await fetch_weather("Paris")
    print(f"First call (live): {result}")

    # Second call is served from cache — <5ms latency, 0 API calls
    result = await fetch_weather("Paris")
    print(f"Second call (cached): {result}")

    # Write operation with circuit breaker protection
    await send_alert("Agent completed successfully.")

asyncio.run(main())

---

🧠 Core Concepts

1. Cache Backends

Register backends once at application startup, then reference them by name. ToolOps supports multiple backends simultaneously.

from toolops import cache_manager
from toolops.cache import (
    MemoryCache,
    FileCache,
    PostgresCache,
    SQLiteCache,
    ValkeyCache,
    RedisCache,
    MySQLCache,
    SemanticCache,
    SentenceTransformerEmbedder,
)


# In-memory: fastest, cleared on restart, no extra dependencies
cache_manager.register("memory", MemoryCache(), is_default=True)


# File: zero-dependency persistent cache, ideal for single-process apps
cache_manager.register("file", FileCache("/tmp/toolops-cache"))


# SQLite: lightweight persistent cache, single-file, no server required
cache_manager.register("sqlite", SQLiteCache("toolops_cache.db"))


# Postgres: persistent across restarts, shareable across processes
cache_manager.register("db", PostgresCache("postgresql://user:pass@localhost:5432/mydb"))


# Valkey / Redis: distributed in-memory cache with async pooling
cache_manager.register("valkey", ValkeyCache(host="localhost", port=6379))
cache_manager.register("redis", RedisCache(url="redis://localhost:6379/0"))


# MySQL / MariaDB: persistent relational cache
cache_manager.register("mysql", MySQLCache(host="localhost", db="myapp", user="root", password="secret"))
# — or via DSN —
cache_manager.register("mysql", MySQLCache(dsn="mysql://root:secret@localhost:3306/myapp"))


# Semantic: vector embeddings to match by meaning, not string equality
# Reduces LLM calls up to 90%
embedder = SentenceTransformerEmbedder("all-MiniLM-L6-v2")
cache_manager.register("semantic", SemanticCache(embedder=embedder, threshold=0.92))

2. Resilience Patterns

ToolOps provides robust, battle-tested resilience out of the box.

• Circuit Breaker: Prevents hammering a failing service and causing cascading failures.
• Stale-if-Error: Serves the last known good cached value if the live API call fails.
• Request Coalescing: If 50 agents call the same endpoint simultaneously, ToolOps executes the real API call once and multicasts the result.

@readonly(
    cache_backend   = "db",
    cache_ttl       = 3600,
    retry_count     = 3,
    timeout         = 10.0,
    stale_if_error  = True,     # Fallback on API failure
    circuit_breaker = True      # Protect the underlying service
)
async def get_market_data(ticker: str) -> dict:
    return await api.fetch(ticker)

3. Architecture & Security (v1.0.0)

ToolOps v1.0.0 introduces an enterprise-grade architecture:

• Middleware Pipeline: The monolithic decorator has been refactored into a composable pipeline (Logging, Cache, CircuitBreaker, Retry, Coalescing, Fallback).
• SHA-256 Cache Key Hashing: All cache keys are strictly hashed. No sensitive data (tokens, PII) is exposed in cache stores.
• Automatic Parameter Masking: Tool arguments containing sensitive keywords (token, password, secret, etc.) are automatically masked as ***MASKED*** in structured logs.

---

📊 Observability

ToolOps instruments every tool call automatically.

OpenTelemetry (OTEL) & Prometheus

from toolops import configure_opentelemetry, prometheus_metrics

# 1. Configure OpenTelemetry tracing (accepts any standard tracer instance)
configure_opentelemetry(tracer)


# 2. Expose Prometheus metrics (returns a raw Prometheus text string)
metrics_string = prometheus_metrics()

Key metrics exposed include toolops_cache_hits_total, toolops_tool_latency_seconds, and toolops_circuit_opens_total.

---

🔌 Framework Integration

ToolOps decorates plain Python async functions, making it 100% compatible with your favorite agent frameworks.

LangChain / LangGraph

from langchain.tools import tool

@tool
@readonly(cache_backend="memory", cache_ttl=600)
async def search_web(query: str) -> str:
    """Search the web and return a summary."""
    return await web_search_api.run(query)

CrewAI

from crewai.tools import BaseTool

class ResearchTool(BaseTool):
    name: str = "Research Tool"
    description: str = "Fetches and caches research data."

    @readonly(cache_backend="db", cache_ttl=3600)
    async def _run(self, query: str) -> str:
        return await research_api.fetch(query)

Model Context Protocol (MCP)

from toolops.integrations.mcp import MCPIntegration

# Generate a fully typed MCP tool definition automatically
mcp_definition = MCPIntegration.to_mcp_definition(get_weather)
mcp_server.register_tool(mcp_definition)

---

🛠️ CLI Reference

ToolOps ships with a command-line tool for managing your cache infrastructure.

# Check the health of all registered backends
toolops doctor

# View live cache statistics for an app
toolops stats --app my_app:setup_toolops

# Clear a specific backend's cache
toolops clear memory --app my_app:setup_toolops

---

🤝 Contributing

ToolOps is built for the community, by the community.

• Review our Contributing Guide to get started.
• Check out the Code of Conduct.
• Report security issues safely via our Security Policy.

---

💬 Community & Contact

We are actively building the future of AI Agent infrastructure. Join the discussion!

• Creator: Hedi Manai (LinkedIn | GitHub)
• Report Bugs & Feature Requests: GitHub Issues
• Email: [e-mail]

---

ToolOps — Built for Production.
Licensed under Apache 2.0