Skip to main content

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

ScenarioRecommended Approach
Simple lookupDirect tool call
Batch operationsDirect tool call
External integrationDirect tool call
Scheduled tasksDirect tool call
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']}")
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"
}
}
}
)
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")

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']}")
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

  1. Validate Parameters: Check inputs before sending
  2. Handle Timeouts: Set appropriate timeouts
  3. Retry Transient Failures: Implement retry logic
  4. Log Errors: Track and monitor tool failures

Performance

  1. Batch Operations: Execute multiple tools in parallel when independent
  2. Cache Results: Cache tool outputs when appropriate
  3. Optimize Parameters: Tune for your use case
  4. Monitor Costs: Track web search and code execution usage

Security

  1. Sanitize Inputs: Validate user inputs
  2. Validate Permissions: Ensure API key has tool permissions
  3. Sandbox Code: Code runs in isolated environment
  4. 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
}

Next Steps

  • Tools - Learn about all available tools
  • Agents - Use tools with agent orchestration
  • Examples - See more patterns