Python Tool Calling Examples

Clean, production-ready examples for using tool calling with Captain API in Python.

Basic Calculator Tool

from openai import OpenAI
import json

client = OpenAI(
    base_url="https://api.runcaptain.com/v1",
    api_key="your_api_key",
    default_headers={"X-Organization-ID": "your_org_id"}
)

def calculate(operation, a, b):
    operations = {
        "add": lambda x, y: x + y,
        "subtract": lambda x, y: x - y,
        "multiply": lambda x, y: x * y,
        "divide": lambda x, y: x / y if y != 0 else None
    }
    return operations[operation](a, b)

tools = [{
    "type": "function",
    "function": {
        "name": "calculate",
        "description": "Perform arithmetic calculations",
        "parameters": {
            "type": "object",
            "properties": {
                "operation": {
                    "type": "string",
                    "enum": ["add", "subtract", "multiply", "divide"]
                },
                "a": {"type": "number"},
                "b": {"type": "number"}
            },
            "required": ["operation", "a", "b"]
        },
        "strict": True
    }
}]

response = client.chat.completions.create(
    model="captain-voyager-latest",
    messages=[
        {"role": "user", "content": "What is 156 times 243?"}
    ],
    tools=tools
)

if response.choices[0].finish_reason == "tool_calls":
    tool_call = response.choices[0].message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    result = calculate(args["operation"], args["a"], args["b"])
    print(f"Result: {result}")

Weather API Tool

from openai import OpenAI
import json
import requests

client = OpenAI(
    base_url="https://api.runcaptain.com/v1",
    api_key="your_api_key",
    default_headers={"X-Organization-ID": "your_org_id"}
)

def get_weather(location, unit="celsius"):
    api_key = "your_weather_api_key"
    url = f"https://api.weatherapi.com/v1/current.json"
    params = {"key": api_key, "q": location}

    response = requests.get(url, params=params)
    data = response.json()

    temp = data["current"]["temp_c"] if unit == "celsius" else data["current"]["temp_f"]

    return {
        "location": location,
        "temperature": temp,
        "unit": unit,
        "condition": data["current"]["condition"]["text"]
    }

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current weather for a location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City name"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature unit"
                }
            },
            "required": ["location"]
        },
        "strict": True
    }
}]

response = client.chat.completions.create(
    model="captain-voyager-latest",
    messages=[
        {"role": "user", "content": "What's the weather in Tokyo?"}
    ],
    tools=tools
)

if response.choices[0].finish_reason == "tool_calls":
    tool_call = response.choices[0].message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    weather = get_weather(args["location"], args.get("unit", "celsius"))
    print(f"Weather in {weather['location']}: {weather['temperature']}°{weather['unit'][0].upper()}, {weather['condition']}")

Database Query Tool

from openai import OpenAI
import json
import sqlite3

client = OpenAI(
    base_url="https://api.runcaptain.com/v1",
    api_key="your_api_key",
    default_headers={"X-Organization-ID": "your_org_id"}
)

def query_database(customer_id):
    conn = sqlite3.connect('customers.db')
    cursor = conn.cursor()

    cursor.execute(
        "SELECT name, email, phone FROM customers WHERE id = ?",
        (customer_id,)
    )

    result = cursor.fetchone()
    conn.close()

    if result:
        return {
            "name": result[0],
            "email": result[1],
            "phone": result[2]
        }
    return {"error": "Customer not found"}

tools = [{
    "type": "function",
    "function": {
        "name": "query_database",
        "description": "Look up customer information by ID",
        "parameters": {
            "type": "object",
            "properties": {
                "customer_id": {
                    "type": "string",
                    "description": "Customer ID to look up"
                }
            },
            "required": ["customer_id"]
        },
        "strict": True
    }
}]

response = client.chat.completions.create(
    model="captain-voyager-latest",
    messages=[
        {"role": "user", "content": "Get contact info for customer CUST-12345"}
    ],
    tools=tools
)

if response.choices[0].finish_reason == "tool_calls":
    tool_call = response.choices[0].message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    customer_info = query_database(args["customer_id"])

    if "error" not in customer_info:
        print(f"Customer: {customer_info['name']}")
        print(f"Email: {customer_info['email']}")
        print(f"Phone: {customer_info['phone']}")

Tool Calling with Large Context

from openai import OpenAI
import json

client = OpenAI(
    base_url="https://api.runcaptain.com/v1",
    api_key="your_api_key",
    default_headers={"X-Organization-ID": "your_org_id"}
)

# Read large financial report
with open('financial_report.txt', 'r') as f:
    report_content = f.read()

def calculate_total(numbers):
    return {"total": sum(numbers)}

def calculate_average(numbers):
    return {"average": sum(numbers) / len(numbers) if numbers else 0}

tools = [
    {
        "type": "function",
        "function": {
            "name": "calculate_total",
            "description": "Calculate the sum of a list of numbers",
            "parameters": {
                "type": "object",
                "properties": {
                    "numbers": {
                        "type": "array",
                        "items": {"type": "number"},
                        "description": "List of numbers to sum"
                    }
                },
                "required": ["numbers"]
            },
            "strict": True
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate_average",
            "description": "Calculate the average of a list of numbers",
            "parameters": {
                "type": "object",
                "properties": {
                    "numbers": {
                        "type": "array",
                        "items": {"type": "number"},
                        "description": "List of numbers"
                    }
                },
                "required": ["numbers"]
            },
            "strict": True
        }
    }
]

response = client.chat.completions.create(
    model="captain-voyager-latest",
    messages=[
        {"role": "user", "content": "What is the total revenue across all quarters?"}
    ],
    tools=tools,
    extra_body={
        "captain": {
            "context": report_content
        }
    }
)

if response.choices[0].finish_reason == "tool_calls":
    tool_call = response.choices[0].message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)

    if tool_call.function.name == "calculate_total":
        result = calculate_total(args["numbers"])
        print(f"Total Revenue: ${result['total']:,.2f}")
    elif tool_call.function.name == "calculate_average":
        result = calculate_average(args["numbers"])
        print(f"Average Revenue: ${result['average']:,.2f}")

Error Handling

from openai import OpenAI
import json

client = OpenAI(
    base_url="https://api.runcaptain.com/v1",
    api_key="your_api_key",
    default_headers={"X-Organization-ID": "your_org_id"}
)

def safe_execute_tool(tool_name, args):
    try:
        if tool_name == "calculate":
            return calculate(args["operation"], args["a"], args["b"])
        elif tool_name == "query_database":
            return query_database(args["customer_id"])
        elif tool_name == "get_weather":
            return get_weather(args["location"], args.get("unit", "celsius"))
        else:
            return {"error": f"Unknown tool: {tool_name}"}
    except KeyError as e:
        return {"error": f"Missing parameter: {e}"}
    except Exception as e:
        return {"error": f"Tool execution failed: {str(e)}"}

tools = [...]  # Your tool definitions

response = client.chat.completions.create(
    model="captain-voyager-latest",
    messages=[{"role": "user", "content": "Your query"}],
    tools=tools
)

if response.choices[0].finish_reason == "tool_calls":
    for tool_call in response.choices[0].message.tool_calls:
        try:
            args = json.loads(tool_call.function.arguments)
        except json.JSONDecodeError:
            print(f"Invalid JSON in tool arguments: {tool_call.function.arguments}")
            continue

        result = safe_execute_tool(tool_call.function.name, args)

        if "error" in result:
            print(f"Tool error: {result['error']}")
        else:
            print(f"Tool result: {result}")

Multiple Tools

from openai import OpenAI
import json

client = OpenAI(
    base_url="https://api.runcaptain.com/v1",
    api_key="your_api_key",
    default_headers={"X-Organization-ID": "your_org_id"}
)

# Define multiple tools
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "Get current stock price",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {"type": "string", "description": "Stock ticker symbol"}
                },
                "required": ["symbol"]
            },
            "strict": True
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate_roi",
            "description": "Calculate return on investment",
            "parameters": {
                "type": "object",
                "properties": {
                    "initial": {"type": "number", "description": "Initial investment"},
                    "current": {"type": "number", "description": "Current value"}
                },
                "required": ["initial", "current"]
            },
            "strict": True
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_company_info",
            "description": "Get company information",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {"type": "string", "description": "Stock ticker symbol"}
                },
                "required": ["symbol"]
            },
            "strict": True
        }
    }
]

response = client.chat.completions.create(
    model="captain-voyager-latest",
    messages=[
        {"role": "user", "content": "What's the current price of AAPL and its ROI if I bought at $150?"}
    ],
    tools=tools
)

if response.choices[0].finish_reason == "tool_calls":
    for tool_call in response.choices[0].message.tool_calls:
        args = json.loads(tool_call.function.arguments)

        if tool_call.function.name == "get_stock_price":
            price = get_stock_price(args["symbol"])
            print(f"Current price of {args['symbol']}: ${price}")

        elif tool_call.function.name == "calculate_roi":
            roi = calculate_roi(args["initial"], args["current"])
            print(f"ROI: {roi}%")

        elif tool_call.function.name == "get_company_info":
            info = get_company_info(args["symbol"])
            print(f"Company: {info['name']}, Sector: {info['sector']}")

Async Tool Execution

from openai import AsyncOpenAI
import json
import asyncio
import aiohttp

client = AsyncOpenAI(
    base_url="https://api.runcaptain.com/v1",
    api_key="your_api_key",
    default_headers={"X-Organization-ID": "your_org_id"}
)

async def fetch_data_async(endpoint):
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/{endpoint}") as response:
            return await response.json()

tools = [{
    "type": "function",
    "function": {
        "name": "fetch_data",
        "description": "Fetch data from API endpoint",
        "parameters": {
            "type": "object",
            "properties": {
                "endpoint": {"type": "string", "description": "API endpoint to call"}
            },
            "required": ["endpoint"]
        },
        "strict": True
    }
}]

async def main():
    response = await client.chat.completions.create(
        model="captain-voyager-latest",
        messages=[
            {"role": "user", "content": "Fetch user data from /users/123"}
        ],
        tools=tools
    )

    if response.choices[0].finish_reason == "tool_calls":
        tool_call = response.choices[0].message.tool_calls[0]
        args = json.loads(tool_call.function.arguments)

        data = await fetch_data_async(args["endpoint"])
        print(f"Fetched data: {data}")

asyncio.run(main())