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

PyPI version Python Tests Coverage PyPI Downloads License GitHub Stars

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

Website · Documentation · Quickstart · Changelog

⚡ 30-Second Pitch

# 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

⚖️ Why ToolOps?

📦 Installation

pip install toolops

💻 OS-Specific Guides

🐧 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

# 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

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

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

📊 Observability

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

🔌 Framework Integration

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

# 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

💬 Community & Contact