Direct Tool Calling
Execute specific tools without agent orchestration.
Overview
Direct tool calling bypasses the agent's planning logic and executes tools immediately.
When to Use Direct Calling
| Scenario | Recommended Approach |
|---|---|
| Simple lookup | Direct tool call |
| Batch operations | Direct tool call |
| External integration | Direct tool call |
| Scheduled tasks | Direct tool call |
RAG Search
Basic Search
import requests
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"tool_name": "rag_search",
"parameters": {
"query": "API key configuration",
"collection_name": "documentation",
"top_k": 5,
"similarity_threshold": 0.7
}
}
)
results = response.json()["results"]
for i, result in enumerate(results, 1):
print(f"{i}. Score: {result['score']:.4f}")
print(f" Content: {result['content'][:100]}...")
print(f" Document: {result['metadata']['document_id']}")
Parameterized Search
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Execution": "Bearer YOUR_API_KEY"},
json={
"tool_name": "rag_search",
"parameters": {
"query": "API configuration",
"collection_name": "documentation",
"top_k": 10, # More results
"similarity_threshold": 0.6, # Lower threshold
"filters": { # Optional filters
"document_type": "policy",
"created_after": "2024-01-01"
}
}
}
)
Batch Search
import requests
queries = [
"API configuration",
"Reset password",
"Update settings",
"Billing information"
]
results = []
for query in queries:
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"tool_name": "rag_search",
"parameters": {
"query": query,
"collection_name": "documentation",
"top_k": 3
}
}
)
results.extend(response.json()["results"])
# Process all results
print(f"Processed {len(results)} queries")
Web Search
Basic Search
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"tool_name": "web_search",
"parameters": {
"query": "Latest AI developments 2024",
"max_results": 10
}
}
)
results = response.json()["results"]
for i, result in enumerate(results, 1):
print(f"{i}. {result['title']}")
print(f" URL: {result['url']}")
print(f" Published: {result['published_at']}")
Time-Bound Search
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"tool_name": "web_search",
"parameters": {
"query": "AI news last 7 days",
"max_results": 20,
"filters": {
"published_after": (datetime.now() - timedelta(days=7)).isoformat()
}
}
}
)
News Search by Topic
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"tool_name": "web_search",
"parameters": {
"query": "artificial intelligence breakthroughs 2024",
"max_results": 15,
"filters": {
"site": "techcrunch.com" # Specific site
}
}
}
)
Code Execution
Simple Python Script
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"tool_name": "code_execution",
"parameters": {
"code": """
import pandas as pd
import numpy as np
# Load data
data = pd.read_csv('sales.csv')
# Calculate mean
mean = data['sales'].mean()
print(f"Mean: {mean:.2f}")
"""
}
}
)
result = response.json()
print(f"Output: {result['output']}")
print(f"Stderr: {result.get('stderr', '')}")
Data Analysis
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"tool_name": "code_execution",
"parameters": {
"code": """
import pandas as pd
from sklearn.linear_model import LinearRegression
import numpy as np
# Prepare data
data = pd.read_csv('data.csv')
X = data[['feature1', 'feature2', 'feature3']]
y = data['target']
# Train model
model = LinearRegression()
model.fit(X, y)
# Predict
predictions = model.predict(X_new)
"""
}
}
)
result = response.json()
print(f"Model coefficients: {result['output']}")
Visualization
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"tool_name": "code_execution",
"parameters": {
"code": """
import matplotlib.pyplot as plt
import pandas as pd
# Load data
data = pd.read_csv('data.csv')
# Create plot
plt.figure(figsize=(10, 6))
plt.plot(data['date'], data['value'])
plt.xlabel('Date')
plt.ylabel('Value')
plt.title('Data Trend')
plt.xticks(rotation=45)
# Return as base64
import base64
from io import BytesIO
buf = BytesIO()
plt.savefig(buf, format='png')
plot_data = base64.b64encode(buf.getvalue()).decode()
"""
}
}
)
result = response.json()
print(f"Plot data: {result['output']}")
MCP Tools
Execute custom MCP tools directly.
Order Lookup
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"tool_name": "mcp_order_api",
"parameters": {
"order_id": "12345",
"action": "get_status"
}
}
)
result = response.json()
print(f"Order Status: {result['result']['status']}")
print(f"Customer: {result['result']['customer']['name']}")
Weather Information
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"tool_name": "mcp_weather",
"parameters": {
"location": "San Francisco, CA"
}
}
)
result = response.json()
print(f"Temperature: {result['result']['temperature']}°F")
print(f"Conditions: {result['result']['conditions']}")
Custom Business Logic
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"tool_name": "mcp_erp_system",
"parameters": {
"action": "get_customer_balance",
"customer_id": "CUST-12345"
}
}
)
result = response.json()
print(f"Balance: ${result['result']['balance']:.2f}")
print(f"Credit Limit: {result['result']['credit_limit']:.2f}")
Error Handling
Common Tool Errors
Tool Not Found
{
"error": "TOOL_NOT_FOUND",
"message": "Tool 'custom_tool' is not available"
}
Invalid Parameters
{
"error": "INVALID_PARAMETERS",
"message": "Parameter 'timeout' must be positive number"
}
Execution Failed
{
"error": "TOOL_EXECUTION_FAILED",
"message": "Tool execution returned an error",
"details": {
"error_type": "timeout",
"tool_name": "code_execution"
}
}
Best Practices
Error Handling
- Validate Parameters: Check inputs before sending
- Handle Timeouts: Set appropriate timeouts
- Retry Transient Failures: Implement retry logic
- Log Errors: Track and monitor tool failures
Performance
- Batch Operations: Execute multiple tools in parallel when independent
- Cache Results: Cache tool outputs when appropriate
- Optimize Parameters: Tune for your use case
- Monitor Costs: Track web search and code execution usage
Security
- Sanitize Inputs: Validate user inputs
- Validate Permissions: Ensure API key has tool permissions
- Sandbox Code: Code runs in isolated environment
- Audit MCP Tools: Regularly review custom tool access
Use Cases
Document Search API
# Build a document search service
def search_documents(query, top_k=5):
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"tool_name": "rag_search",
"parameters": {
"query": query,
"collection_name": "documentation",
"top_k": top_k
}
}
)
return response.json()["results"]
# Usage
results = search_documents("API configuration")
for result in results:
print(f"Score: {result['score']:.4f}")
print(f"Content: {result['content'][:100]}...")
Scheduled Data Analysis
import requests
import schedule
def daily_analysis():
# Execute code analysis
response = requests.post(
"http://localhost/api/v1/tools/execute",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"tool_name": "code_execution",
"parameters": {
"code": """
import pandas as pd
data = pd.read_csv('daily_sales.csv')
daily_summary = data.groupby('date').agg({'sales': 'sum'})
print(daily_summary.tail())
"""
}
}
)
return response.json()["output"]
# Schedule to run daily at 8 AM
schedule.every().day.at("08:00").do(daily_analysis)
Multi-Tool Workflows
Sequential Tool Execution
def multi_step_analysis(query):
# Step 1: RAG search
rag_result = execute_tool("rag_search", {
"query": query,
"collection_name": "docs",
"top_k": 5
})
# Step 2: Web search
web_result = execute_tool("web_search", {
"query": f"{query} latest developments"
})
# Step 3: Code analysis with combined context
context = f"RAG: {rag_result}\nWeb: {web_result}"
analysis_code = f"""
# Combine results from RAG and web search
rag_data = {item['content'] for item in rag_result}
web_data = web_result
# Generate analysis
print(f"Analyzing with context: {context}")
"""
# Step 4: Execute analysis
analysis_result = execute_tool("code_execution", {
"code": analysis_code
})
return {
"rag_result": rag_result,
"web_result": web_result,
"analysis_result": analysis_result
}