Tool Calling

Tool Calling

Overview

Who is this for? Developers building AI applications that need to integrate with external APIs, databases, or services through structured function calls.

What you'll achieve: Enable AI models to call functions, retrieve real-time data, perform calculations, and interact with external systems in a controlled, type-safe manner across multiple providers.

Tool calling allows AI models to execute predefined functions with structured parameters, enabling dynamic interactions with external systems, APIs, and data sources.

Supported Providers

ProviderFunction CallingParallel CallsStreamingTool Choice
OpenAI
OpenAI-Compatible
Anthropic Claude
Google AI (Gemini)
Groq
Cohere

Basic Tool Calling

Single Function Definition

<CODE_PLACEHOLDER>
Response:
<CODE_PLACEHOLDER>

Function Execution and Follow-up

<CODE_PLACEHOLDER>

Advanced Tool Calling

Multiple Tools Definition

<CODE_PLACEHOLDER>

Parallel Function Calls

<CODE_PLACEHOLDER>

Tool Choice Control

<CODE_PLACEHOLDER>

Streaming Tool Calls

Streaming Function Arguments

<CODE_PLACEHOLDER>
Streaming Response:

<CODE_PLACEHOLDER>

Implementation Examples

Node.js Tool Execution Handler

<CODE_PLACEHOLDER>

Python Async Tool Handler

<CODE_PLACEHOLDER>
import json
import asyncio
from typing import Dict, Any, Callable, List
import aiohttp

class AsyncToolExecutor:
def init(self):
self.tools: Dict[str, Dict[str, Any]] = {}
self.register_builtin_tools()

def register_tool(self, name: str, fn: Callable, schema: Dict[str, Any]):
    self.tools[name] = {'fn': fn, 'schema': schema}

def register_builtin_tools(self):
    # Database query tool
    async def query_database(params: Dict[str, Any]) -> Dict[str, Any]:
        # Simulate database query
        await asyncio.sleep(0.1)
        return {
            "results": f"Query '{params['query']}' returned 5 rows",
            "count": 5
        }
    
    self.register_tool('query_database', query_database, {
        'name': 'query_database',
        'description': 'Execute SQL queries on the database',
        'parameters': {
            'type': 'object',
            'properties': {
                'query': {'type': 'string'},
                'limit': {'type': 'integer', 'default': 10}
            },
            'required': ['query']
        }
    })
    
    # HTTP API call tool
    async def http_request(params: Dict[str, Any]) -> Dict[str, Any]:
        async with aiohttp.ClientSession() as session:
            async with session.get(params['url']) as response:
                return {
                    'status': response.status,
                    'data': await response.text()
                }
    
    self.register_tool('http_request', http_request, {
        'name': 'http_request',
        'description': 'Make HTTP GET requests',
        'parameters': {
            'type': 'object',
            'properties': {
                'url': {'type': 'string', 'format': 'uri'}
            },
            'required': ['url']
        }
    })

async def execute_tool(self, tool_call: Dict[str, Any]) -> Dict[str, Any]:
    name = tool_call['function']['name']
    args_str = tool_call['function']['arguments']
    
    if name not in self.tools:
        return {'error': f"Tool '{name}' not found"}
    
    try:
        args = json.loads(args_str)
        result = await self.tools[name]['fn'](args)
        return {'result': result}
    except Exception as e:
        return {'error': str(e)}

def get_tool_schemas(self) -> List[Dict[str, Any]]:
    return [
        {'type': 'function', 'function': tool['schema']}
        for tool in self.tools.values()
    ]

Usage example

async def chat_with_async_tools():
executor = AsyncToolExecutor()

messages = [{"role": "user", "content": "Query the users table for active users"}]

async with aiohttp.ClientSession() as session:
    async with session.post(
        'https://api.orq.ai/v2/chat/completions',
        headers={'Authorization': f'Bearer {os.getenv("ORQ_API_KEY")}'},
        json={
            'model': 'gpt-4',
            'messages': messages,
            'tools': executor.get_tool_schemas(),
        }
    ) as response:
        data = await response.json()
        
    message = data['choices'][0]['message']
    
    if message.get('tool_calls'):
        # Execute tools concurrently
        tasks = [
            executor.execute_tool(tool_call) 
            for tool_call in message['tool_calls']
        ]
        results = await asyncio.gather(*tasks)
        
        # Add tool results to conversation
        messages.append(message)
        for tool_call, result in zip(message['tool_calls'], results):
            messages.append({
                'role': 'tool',
                'tool_call_id': tool_call['id'],
                'content': json.dumps(result)
            })

React Hook for Tool Calling

<CODE_PLACEHOLDER>

Provider-Specific Features

OpenAI & Compatible Providers

  • Parallel Functions: Execute multiple tools simultaneously
  • Streaming Support: Stream function arguments as they're generated
  • Tool Choice: Force, disable, or auto-select tool usage
  • Function Validation: Automatic parameter validation against schema

Anthropic Claude

  • Tool Use: Comprehensive function calling with reasoning
  • Multiple Tools: Supports complex multi-step workflows
  • Error Handling: Graceful handling of invalid function calls
  • Streaming: Real-time tool argument generation

Google AI (Gemini)

  • Function Calling: Native function execution capabilities
  • Code Execution: Built-in code interpreter for calculations
  • Multi-Turn: Maintains context across tool interactions

Best Practices

Function Definition Guidelines

<CODE_PLACEHOLDER>

Error Handling

<CODE_PLACEHOLDER>

Security Considerations

  • Input Validation: Always validate function arguments
  • Rate Limiting: Implement per-user tool execution limits
  • Permissions: Check user permissions before tool execution
  • Sanitization: Sanitize inputs for database queries and API calls

Troubleshooting

Common Issues

Invalid Function Arguments
<CODE_PLACEHOLDER>
Function Not Found
<CODE_PLACEHOLDER>
Execution Timeout
<CODE_PLACEHOLDER>

Next Steps