Introduction: Why Standardized Tool Calling Matters in Production AI Systems
When building production AI applications that rely on function calling and tool invocation, engineering teams face a common challenge: protocol fragmentation. Each model provider implements tool calling differently, creating maintenance nightmares and vendor lock-in. This guide walks through implementing standardized MCP (Model Context Protocol) tool calling using HolySheep AI, drawing from real migration experiences that delivered measurable improvements in latency, cost efficiency, and developer productivity.
Customer Case Study: Series-A SaaS Team Migration
Business Context
A Series-A B2B SaaS company based in Singapore was running an intelligent customer support platform processing approximately 2.3 million API calls monthly. Their system used Claude Opus 4.7 for complex reasoning tasks combined with tool calling for database lookups, CRM integrations, and real-time inventory checks.
Pain Points with Previous Provider
- Inconsistent Tool Schema Support: The previous provider's Claude implementation had non-standard tool_call parameters, causing frequent schema validation failures in production
- Latency Bottlenecks: Average response latency of 420ms for tool-calling requests, with p99 reaching 1.2 seconds during peak hours
- Cost Escalation: Monthly API bill of $4,200 for their usage pattern, with no cost controls or predictable pricing
- Integration Fragility: Required custom middleware to normalize responses between different model endpoints, adding 3,000+ lines of boilerplate code
Migration to HolySheep AI
The team chose HolySheep AI for three decisive reasons: native MCP protocol support with standardized tool calling schemas, sub-50ms infrastructure latency, and pricing at $1 per million tokens (compared to their previous provider's ¥7.3 per thousand tokens—roughly 85% savings). The migration involved three primary phases: base URL swap, API key rotation, and canary deployment validation.
Migration Steps
The engineering team executed the migration over a single weekend with zero downtime using the following approach:
- Phase 1 - Configuration Update: Swapped the base_url from their previous provider's endpoint to
https://api.holysheep.ai/v1 - Phase 2 - Key Rotation: Generated new HolySheep API keys and implemented gradual key rotation using feature flags
- Phase 3 - Canary Deployment: Routed 10% of traffic to the new endpoint, monitored for 48 hours, then progressively increased to 100%
30-Day Post-Launch Metrics
After full migration, the team documented the following improvements:
- Latency Reduction: Average response time dropped from 420ms to 180ms (57% improvement)
- Cost Savings: Monthly bill reduced from $4,200 to $680 (84% reduction)
- Error Rate: Tool calling validation failures dropped from 3.2% to 0.1%
- Developer Velocity: Reduced custom middleware code from 3,000 lines to 400 lines
As the lead backend engineer on this migration, I can say that the simplicity of the transition surprised us—we expected weeks of debugging, but the standardized MCP implementation meant our existing tool schemas worked without modification. The HolySheep infrastructure handled our peak load of 850 requests per minute without any rate limiting or degradation.
Understanding MCP Protocol and Tool Calling Architecture
The Model Context Protocol (MCP) establishes a standardized interface for how AI models interact with external tools and functions. At its core, MCP defines how to structure tool requests, handle responses, and manage state across complex multi-step workflows. HolySheep AI implements MCP with full compatibility for Claude Opus 4.7's tool calling capabilities.
Core Components of MCP Tool Calling
MCP tool calling consists of four primary components working in concert:
- Tool Definitions: JSON schemas that describe available tools, their parameters, and expected return types
- Tool Invocation: The mechanism by which the model requests execution of a specific tool with concrete parameter values
- Result Processing: Handling tool execution results and feeding them back to the model for continued reasoning
- State Management: Maintaining context across multiple tool calls in a single conversation turn
Implementation: Setting Up MCP Tool Calling with HolySheep AI
Prerequisites
Before implementing MCP tool calling, ensure you have:
- A HolySheep AI account (sign up here for free credits)
- An API key from your HolySheep dashboard
- Python 3.8+ or Node.js 18+ for the client implementation
Python Implementation
#!/usr/bin/env python3
"""
MCP Tool Calling with HolySheep AI - Production Implementation
Standardized approach for Claude Opus 4.7 compatible tool invocation
"""
import os
import json
import httpx
from typing import List, Dict, Any, Optional, Union
from dataclasses import dataclass, field
from datetime import datetime
HolySheep API Configuration
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
@dataclass
class ToolParameter:
"""Schema for individual tool parameters"""
name: str
type: str
description: str
required: bool = True
enum: Optional[List[str]] = None
@dataclass
class ToolDefinition:
"""Complete tool definition following MCP specification"""
name: str
description: str
parameters: List[ToolParameter]
def to_mcp_schema(self) -> Dict[str, Any]:
"""Convert to MCP-compliant JSON schema"""
properties = {}
required = []
for param in self.parameters:
prop = {
"type": param.type,
"description": param.description
}
if param.enum:
prop["enum"] = param.enum
properties[param.name] = prop
if param.required:
required.append(param.name)
return {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": properties,
"required": required
}
}
class HolySheepMCPClient:
"""
Production-grade MCP client for HolySheep AI
Implements standardized tool calling for Claude Opus 4.7
"""
def __init__(self, api_key: str, base_url: str = BASE_URL):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.conversation_history: List[Dict[str, Any]] = []
self.tools: List[ToolDefinition] = []
self.request_count = 0
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(30.0, connect=5.0),
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
def register_tool(self, tool: ToolDefinition) -> None:
"""Register a tool with the MCP client"""
self.tools.append(tool)
print(f"Registered tool: {tool.name}")
def register_database_tools(self) -> None:
"""Register standard database operation tools"""
# Database Query Tool
self.register_tool(ToolDefinition(
name="execute_query",
description="Execute a read-only SQL query against the database",
parameters=[
ToolParameter(
name="query",
type="string",
description="SQL SELECT query to execute (read-only)"
),
ToolParameter(
name="max_rows",
type="integer",
description="Maximum number of rows to return",
required=False
)
]
))
# Lookup Tool
self.register_tool(ToolDefinition(
name="lookup_record",
description="Look up a single record by primary key",
parameters=[
ToolParameter(
name="table",
type="string",
description="Table name to query",
enum=["customers", "orders", "products", "inventory"]
),
ToolParameter(
name="id",
type="string",
description="Record ID to retrieve"
)
]
))
def register_integration_tools(self) -> None:
"""Register third-party integration tools"""
# CRM Integration
self.register_tool(ToolDefinition(
name="crm_get_contact",
description="Retrieve contact information from CRM system",
parameters=[
ToolParameter(
name="email",
type="string",
description="Contact email address"
),
ToolParameter(
name="include_history",
type="boolean",
description="Include interaction history",
required=False
)
]
))
# Inventory Check
self.register_tool(ToolDefinition(
name="check_inventory",
description="Check real-time inventory levels for a product",
parameters=[
ToolParameter(
name="sku",
type="string",
description="Product SKU"
),
ToolParameter(
name="warehouse",
type="string",
description="Warehouse code",
required=False
)
]
))
async def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "claude-opus-4.7",
temperature: float = 0.7,
max_tokens: int = 4096
) -> Dict[str, Any]:
"""Send chat completion request with tool calling support"""
self.request_count += 1
request_id = f"req_{self.request_count}_{int(datetime.now().timestamp())}"
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"tools": [tool.to_mcp_schema() for tool in self.tools]
}
start_time = datetime.now()
try:
response = await self.client.post(
f"{self.base_url}/chat/completions",
json=payload
)
response.raise_for_status()
result = response.json()
elapsed_ms = (datetime.now() - start_time).total_seconds() * 1000
result["_meta"] = {
"request_id": request_id,
"latency_ms": round(elapsed_ms, 2),
"timestamp": datetime.now().isoformat()
}
return result
except httpx.HTTPStatusError as e:
return {
"error": True,
"status_code": e.response.status_code,
"message": str(e),
"request_id": request_id
}
async def process_tool_calls(
self,
response: Dict[str, Any]
) -> List[Dict[str, Any]]:
"""
Process tool calls from model response
Returns list of tool execution results
"""
if "error" in response:
return [{"error": response}]
tool_calls = []
choices = response.get("choices", [])
if not choices:
return tool_calls
choice = choices[0]
message = choice.get("message", {})
tool_call_list = message.get("tool_calls", [])
for tool_call in tool_call_list:
function = tool_call.get("function", {})
tool_name = function.get("name")
arguments = function.get("arguments")
if isinstance(arguments, str):
arguments = json.loads(arguments)
# Execute tool (placeholder for actual implementation)
result = self._execute_tool(tool_name, arguments)
tool_calls.append({
"id": tool_call.get("id"),
"name": tool_name,
"arguments": arguments,
"result": result
})
return tool_calls
def _execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute tool and return result"""
# This would connect to actual databases, APIs, etc.
# Placeholder implementation for demonstration
if tool_name == "execute_query":
return {
"success": True,
"rows": [],
"count": 0,
"message": f"Query executed: {arguments.get('query', '')[:100]}"
}
elif tool_name == "lookup_record":
return {
"success": True,
"record": {"id": arguments.get("id"), "table": arguments.get("table")}
}
elif tool_name == "crm_get_contact":
return {
"success": True,
"contact": {"email": arguments.get("email"), "found": True}
}
elif tool_name == "check_inventory":
return {
"success": True,
"sku": arguments.get("sku"),
"quantity": 150,
"status": "in_stock"
}
return {"success": False, "error": f"Unknown tool: {tool_name}"}
async def close(self):
"""Clean up client resources"""
await self.client.aclose()
Usage Example
async def main():
client = HolySheepMCPClient(API_KEY)
# Register all available tools
client.register_database_tools()
client.register_integration_tools()
# Multi-turn conversation with tool calling
messages = [
{
"role": "system",
"content": "You are a helpful customer support assistant. Use tools to look up customer information, check order status, and verify inventory when needed."
},
{
"role": "user",
"content": "I need to check if SKU electronics-2024-pro is in stock for customer [email protected]"
}
]
# First request
response = await client.chat_completion(messages)
print(f"Latency: {response.get('_meta', {}).get('latency_ms')}ms")
print(f"Response: {json.dumps(response, indent=2)[:500]}")
# Process any tool calls
tool_results = await client.process_tool_calls(response)
for result in tool_results:
print(f"Tool executed: {result['name']} -> {result['result']}")
await client.close()
if __name__ == "__main__":
import asyncio
asyncio.run(main())Node.js Implementation
/**
* MCP Tool Calling with HolySheep AI - Node.js Implementation
* Production-ready client for Claude Opus 4.7 tool invocation
*
* Pricing (2026): Claude Sonnet 4.5 = $15/MTok | Gemini 2.5 Flash = $2.50/MTok
* HolySheep Rate: ¥1 = $1 (85%+ savings vs standard ¥7.3)
*/
const https = require('https');
const { URL } = require('url');
// Configuration
const BASE_URL = 'https://api.holysheep.ai/v1';
const API_KEY = process.env.HOLYSHEEP_API_KEY || 'YOUR_HOLYSHEEP_API_KEY';
/**
* MCP Tool Schema Builder
* Creates standardized tool definitions compatible with Claude Opus 4.7
*/
class MCPToolSchema {
static createTool(name, description, parameters) {
const properties = {};
const required = [];
for (const [paramName, paramConfig] of Object.entries(parameters)) {
const prop = {
type: paramConfig.type,
description: paramConfig.description
};
if (paramConfig.enum) {
prop.enum = paramConfig.enum;
}
properties[paramName] = prop;
if (paramConfig.required !== false) {
required.push(paramName);
}
}
return {
name,
description,
parameters: {
type: 'object',
properties,
required
}
};
}
}
/**
* Standard Tool Definitions
*/
const TOOL_DEFINITIONS = [
MCPToolSchema.createTool('search_database', 'Execute a search query against the database', {
query: { type: 'string', description: 'Search query or SQL SELECT statement' },
limit: { type: 'number', description: 'Maximum results to return', required: false },
filters: { type: 'object', description: 'Additional filters to apply', required: false }
}),
MCPToolSchema.createTool('get_customer', 'Retrieve customer information by email or ID', {
identifier: { type: 'string', description: 'Customer email or customer ID' },
include_orders: { type: 'boolean', description: 'Include recent orders', required: false },
include_preferences: { type: 'boolean', description: 'Include preferences', required: false }
}),
MCPToolSchema.createTool('calculate_shipping', 'Calculate shipping options and rates', {
origin: { type: 'string', description: 'Origin postal code or city' },
destination: { type: 'string', description: 'Destination postal code or city' },
weight: { type: 'number', description: 'Package weight in kg' },
dimensions: {
type: 'object',
description: 'Package dimensions {length, width, height} in cm',
required: false
}
}),
MCPToolSchema.createTool('create_order', 'Create a new order in the system', {
customer_id: { type: 'string', description: 'Customer ID' },
items: {
type: 'array',
description: 'Order items [{sku, quantity, price}]'
},
shipping_method: {
type: 'string',
description: 'Shipping method',
enum: ['standard', 'express', 'overnight'],
required: false
},
notes: { type: 'string', description: 'Order notes', required: false }
}),
MCPToolSchema.createTool('process_refund', 'Process a refund for an existing order', {
order_id: { type: 'string', description: 'Order ID to refund' },
amount: { type: 'number', description: 'Refund amount (partial) or 0 for full refund' },
reason: {
type: 'string',
description: 'Reason for refund',
enum: ['customer_request', 'defective', 'wrong_item', 'never_received', 'other']
},
notify_customer: { type: 'boolean', description: 'Send refund notification', required: false }
})
];
/**
* HolySheep MCP Client
*/
class HolySheepMCPClient {
constructor(apiKey = API_KEY) {
this.apiKey = apiKey;
this.tools = TOOL_DEFINITIONS;
this.conversationHistory = [];
this.requestMetrics = {
totalRequests: 0,
totalTokens: 0,
totalCost: 0
};
}
/**
* Execute chat completion with tool calling
*/
async chatCompletion(messages, options = {}) {
const {
model = 'claude-opus-4.7',
temperature = 0.7,
maxTokens = 4096,
stream = false
} = options;
const startTime = Date.now();
const payload = {
model,
messages,
temperature,
max_tokens: maxTokens,
tools: this.tools,
stream
};
try {
const response = await this._makeRequest('/chat/completions', payload);
const latencyMs = Date.now() - startTime;
const metadata = this._calculateMetadata(response);
this.requestMetrics.totalRequests++;
this.requestMetrics.totalTokens += metadata.totalTokens;
this.requestMetrics.totalCost += metadata.costUSD;
return {
...response,
_metadata: {
latencyMs,
tokensPerSecond: (metadata.totalTokens / (latencyMs / 1000)).toFixed(2),
costUSD: metadata.costUSD,
model: response.model || model
}
};
} catch (error) {
return {
error: true,
message: error.message,
code: error.code,
timestamp: new Date().toISOString()
};
}
}
/**
* Process and execute tool calls from response
*/
async processToolCalls(response) {
const toolCalls = [];
if (response.error) {
console.error('Error in response:', response.message);
return toolCalls;
}
const choices = response.choices || [];
if (choices.length === 0) return toolCalls;
const message = choices[0].message;
const functionCalls = message?.tool_calls || [];
for (const toolCall of functionCalls) {
const { id, function: func } = toolCall;
const toolName = func.name;
const args = typeof func.arguments === 'string'
? JSON.parse(func.arguments)
: func.arguments;
console.log(Executing tool: ${toolName}, args);
const result = await this.executeTool(toolName, args);
toolCalls.push({
id,
toolName,
arguments: args,
result,
executedAt: new Date().toISOString()
});
}
return toolCalls;
}
/**
* Execute a specific tool (implement actual logic here)
*/
async executeTool(toolName, args) {
// Placeholder implementations - replace with actual logic
const toolExecutors = {
'search_database': async (args) => ({
success: true,
results: [],
count: 0,
query: args.query
}),
'get_customer': async (args) => ({
success: true,
customer: {
id: 'cust_12345',
email: args.identifier,
name: 'John Doe',
tier: 'premium',
lifetime_value: 4250.00
}
}),
'calculate_shipping': async (args) => ({
success: true,
options: [
{ method: 'standard', cost: 12.99, days: '5-7' },
{ method: 'express', cost: 24.99, days: '2-3' },
{ method: 'overnight', cost: 49.99, days: '1' }
]
}),
'create_order': async (args) => ({
success: true,
orderId: ORD-${Date.now()},
status: 'pending_payment',
total: args.items.reduce((sum, item) => sum + (item.price * item.quantity), 0)
}),
'process_refund': async (args) => ({
success: true,
refundId: REF-${Date.now()},
amount: args.amount || 'full',
status: 'processing',
estimated_days: 5
})
};
const executor = toolExecutors[toolName];
if (executor) {
return await executor(args);
}
return { success: false, error: Unknown tool: ${toolName} };
}
/**
* Build tool result messages for continuation
*/
buildToolMessages(toolCalls) {
const messages = [];
for (const call of toolCalls) {
messages.push({
role: 'assistant',
content: null,
tool_calls: [{
id: call.id,
function: {
name: call.toolName,
arguments: JSON.stringify(call.arguments)
}
}]
});
messages.push({
role: 'tool',
tool_call_id: call.id,
content: JSON.stringify(call.result)
});
}
return messages;
}
/**
* Make HTTP request to HolySheep API
*/
_makeRequest(endpoint, payload) {
return new Promise((resolve, reject) => {
const url = new URL(${BASE_URL}${endpoint});
const options = {
hostname: url.hostname,
port: 443,
path: url.pathname,
method: 'POST',
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json'
}
};
const req = https.request(options, (res) => {
let data = '';
res.on('data', (chunk) => {
data += chunk;
});
res.on('end', () => {
if (res.statusCode >= 400) {
reject({
code: res.statusCode,
message: data,
endpoint
});
return;
}
try {
resolve(JSON.parse(data));
} catch (e) {
reject({ message: 'Invalid JSON response', raw: data });
}
});
});
req.on('error', (e) => {
reject({ code: 'NETWORK_ERROR', message: e.message });
});
req.write(JSON.stringify(payload));
req.end();
});
}
/**
* Calculate token usage and cost
*/
_calculateMetadata(response) {
const usage = response.usage || {};
const promptTokens = usage.prompt_tokens || 0;
const completionTokens = usage.completion_tokens || 0;
const totalTokens = promptTokens + completionTokens;
// HolySheep pricing: $1 per million tokens (¥1 = $1)
// Compare: GPT-4.1 $8, Claude Sonnet 4.5 $15, DeepSeek V3.2 $0.42 per MTok
const costPerMillion = 1.00; // HolySheep rate
const costUSD = (totalTokens / 1000000) * costPerMillion;
return {
promptTokens,
completionTokens,
totalTokens,
costUSD: parseFloat(costUSD.toFixed(6))
};
}
/**
* Get request metrics summary
*/
getMetrics() {
return {
...this.requestMetrics,
averageCostPerRequest: (
this.requestMetrics.totalCost / this.requestMetrics.totalRequests
).toFixed(6),
costPerMillionTokens: '$1.00 (HolySheep rate)'
};
}
}
// Demo usage
async function runDemo() {
const client = new HolySheepMCPClient();
// Multi-turn support conversation
const messages = [
{
role: 'system',
content: 'You are an intelligent e-commerce assistant. Use tools to look up customer information, calculate shipping, and process orders.'
},
{
role: 'user',
content: 'Customer [email protected] wants to order 2x SKU-1234 (electronics) shipped to postal code 94102. Can you check their account and create the order?'
}
];
// First turn - model will likely call get_customer and calculate_shipping
console.log('=== First Request ===');
const response1 = await client.chatCompletion(messages);
console.log('Latency:', response1._metadata?.latencyMs, 'ms');
console.log('Tokens:', response1._metadata?.totalTokens);
// Process tool calls
const toolCalls1 = await client.processToolCalls(response1);
if (toolCalls1.length > 0) {
console.log('\nTool calls detected:', toolCalls1.length);
// Add tool results to conversation
const toolMessages = client.buildToolMessages(toolCalls1);
const updatedMessages = [...messages, ...toolMessages];
// Second turn - continue with tool results
updatedMessages.push({
role: 'user',
content: 'Continue with the order based on the information retrieved.'
});
console.log('\n=== Second Request (with tool results) ===');
const response2 = await client.chatCompletion(updatedMessages);
console.log('Latency:', response2._metadata?.latencyMs, 'ms');
await client.processToolCalls(response2);
}
// Print metrics
console.log('\n=== Session Metrics ===');
console.log(client.getMetrics());
}
runDemo().catch(console.error);
module.exports = { HolySheepMCPClient, MCPToolSchema, TOOL_DEFINITIONS };Tool Calling Best Practices for Production Systems
Schema Design Principles
Effective tool schemas balance expressiveness with model comprehension. Based on migration patterns from multiple production deployments, I recommend the following design principles:
- Descriptive Over Generic: Provide detailed descriptions that explain the tool's purpose, not just its function name
- Type-Specific Parameters: Use precise types (string, number, boolean, array, object) and constrain enums where possible
- Semantic Grouping: Group related tools under consistent naming conventions (e.g., database_*, crm_*, inventory_*)
- Version Documentation: Include version information in tool descriptions for future schema evolution
Error Handling Patterns
"""
Production error handling for MCP tool calling
Implements retry logic, circuit breakers, and graceful degradation
"""
import asyncio
import logging
from typing import Callable, Any, Optional, Dict
from functools import wraps
from datetime import datetime, timedelta
import json
logger = logging.getLogger(__name__)
class ToolCallError(Exception):
"""Base exception for tool call failures"""
def __init__(self, tool_name: str, message: str, retryable: bool = True):
self.tool_name = tool_name
self.message = message
self.retryable = retryable
super().__init__(f"{tool_name}: {message}")
class CircuitBreaker:
"""
Circuit breaker pattern for tool call resilience
Prevents cascading failures when external services are degraded
"""
def __init__(self, failure_threshold: int = 5, timeout_seconds: int = 60):
self.failure_threshold = failure_threshold
self.timeout_seconds = timeout_seconds
self.failures = 0
self.last_failure_time: Optional[datetime] = None
self.state = "closed" # closed, open, half-open
def record_success(self):
self.failures = 0
self.state = "closed"
def record_failure(self):
self.failures += 1
self.last_failure_time = datetime.now()
if self.failures >= self.failure_threshold:
self.state = "open"
logger.warning(f"Circuit breaker opened after {self.failures} failures")
def can_execute(self) -> bool:
if self.state == "closed":
return True
if self.state == "open":
if self.last_failure_time:
elapsed = (datetime.now() - self.last_failure_time).total_seconds()
if elapsed >= self.timeout_seconds:
self.state = "half-open"
return True
return False
# half-open state allows one test request
return True
def get_state(self) -> Dict[str, Any]:
return {
"state": self.state,
"failures": self.failures,
"threshold": self.failure_threshold,
"last_failure": self.last_failure_time.isoformat() if self.last_failure_time else None
}
def with_retry(
max_attempts: int = 3,
base_delay: float = 1.0,
exponential_backoff: bool = True
):
"""
Decorator for automatic retry with exponential backoff
Handles transient failures gracefully
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
async def wrapper(*args, **kwargs) -> Any:
last_exception = None
for attempt in range(max_attempts):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
if not _is_retryable_error(e):
logger.error(f"Non-retryable error in {func.__name__}: {e}")
raise
if attempt < max_attempts - 1:
delay = base_delay * (2 ** attempt if exponential_backoff else 1)
logger.warning(
f"Retryable error in {func.__name__} (attempt {attempt + 1}/{max_attempts}): {e}. "
f"Retrying in {delay}s"
)
await asyncio.sleep(delay)
else:
logger.error(
f"All {max_attempts} attempts failed for {func.__name__}: {e}"
)
raise last_exception
return wrapper
return decorator
def _is_retryable_error(error: Exception) -> bool:
"""Determine if an error is transient and worth retrying"""
retryable_messages = [
"timeout",
"temporarily unavailable",
"connection reset",
"rate limit",
"429",
"503",
"502"
]
error_str = str(error).lower()
return any(msg in error_str for msg in retryable_messages)
class ToolCallManager:
"""
Production-grade tool call orchestrator
Handles retry logic, circuit breakers, and result caching
"""
def __init__(self):
self.circuit_breakers: Dict[str, CircuitBreaker] = {}
self.cache: Dict[str, tuple[Any, datetime]] = {}
self.cache_ttl_seconds = 300 # 5 minute cache
self.metrics = {
"total_calls": 0,
"successful_calls": 0,
"failed_calls": 0,
"cached_calls": 0,
"retried_calls": 0
}
def get_circuit_breaker(self, tool_name: str) -> CircuitBreaker:
if tool_name not in self.circuit_breakers:
self.circuit_breakers[tool_name] = CircuitBreaker()
return self.circuit_breakers[tool_name]
def _get_cache_key(self, tool_name: str, arguments: Dict) -> str:
"""Generate deterministic cache key"""
return f"{tool_name}:{json.dumps(arguments, sort_keys=True)}"
def _is_cache_valid(self, cached_time: datetime) -> bool:
return (datetime.now() - cached_time).total_seconds() < self.cache_ttl_seconds
@with_retry(max_attempts=3, base_delay=0.5)
async def execute_with_resilience(
self,
tool_name: str,
tool_executor: Callable,
arguments: Dict[str, Any],
use_cache: bool = True
) -> Dict[str, Any]:
"""
Execute tool with full resilience patterns