Example Make and Use of Calculator OpenAPI Tool¶
This example demonstrates how to:
- Create a simple calculator API using FastAPI. The API provides basic arithmetic operations such as addition, subtraction, multiplication, and division.
- Use this API service as a
ToolRegistrytool for LLMs to compute the average of metrics from a file.
We will reuse the file from previous calculator examples, concurrent_raw_results.txt.
Step 1: Create the Calculator API¶
First, let's define a simple FastAPI server that provides basic math operations.
from fastapi import FastAPI, HTTPException
app = FastAPI(
title="OpenAPI Calculator",
description="Provides OpenAPI calculator service for addition, subtraction, multiplication, and division.",
version="1.0.0",
)
@app.get("/add", summary="Addition")
def add(a: float, b: float):
"""
Calculate a + b and return the result.
Args:
a (float): The first operand.
b (float): The second operand.
Returns:
dict: A dictionary containing the key "result" with the sum of a and b.
"""
return a + b
@app.get("/subtract", summary="Subtraction")
def subtract(a: float, b: float):
"""
Calculate a - b and return the result.
Args:
a (float): The first operand.
b (float): The second operand.
Returns:
dict: A dictionary containing the key "result" with the difference of a and b.
"""
return a - b
@app.get("/multiply", summary="Multiplication")
def multiply(a: float, b: float):
"""
Calculate a * b and return the result.
Args:
a (float): The first operand.
b (float): The second operand.
Returns:
dict: A dictionary containing the key "result" with the product of a and b.
"""
return a * b
@app.get("/divide", summary="Division")
def divide(a: float, b: float):
"""
Calculate a / b and return the result.
Args:
a (float): The numerator.
b (float): The denominator.
Returns:
dict: A dictionary containing the key "result" with the quotient of a and b.
Raises:
HTTPException: If b is zero.
"""
if b == 0:
raise HTTPException(status_code=400, detail="Divisor cannot be zero")
return a / b
if __name__ == "__main__":
import os
import uvicorn
port = int(os.getenv("PORT", 8000))
uvicorn.run(app, host="0.0.0.0", port=port)
To run the service, you can use the following command:
or
Step 2: Register and Use the OpenAPI Tool¶
API changes
Previously, register_from_openapi requires spec_url and optional base_url. It was designed to be simple. Yet in practice, we found the need for customization in HTTP requests, for example many OpenAPI services requires authentication headers or custom timeouts. Thus we made the following changes:
The register_from_openapi method new requires two parameters:
client_config: Configures the HTTP client (headers, auth, timeout, etc.) using anHttpClientConfigobject, allowing greater flexibility.openapi_spec: The OpenAPI specification loaded asDict[str, Any]using functions likeload_openapi_specorload_openapi_spec_asyncfrom a file path or URL to the service or specification.
We implement using both Cicada MultiModalModel and OpenAI client to showcase different ways to integrate with the tool registry.
Example:
from toolregistry.integrations.openapi import HttpClientConfig, load_openapi_spec
client_config = HttpClientConfig(base_url="http://localhost:8000")
openapi_spec = load_openapi_spec("./openapi_spec.json") # specification at local path
openapi_spec = load_openapi_spec("http://localhost:8000") # URL to service root
openapi_spec = load_openapi_spec("http://localhost:8000/openapi.json") # URL to specification
registry.register_from_openapi(
client_config=client_config,
openapi_spec=openapi_spec,
namespace=False,
)
Cicada MultiModalModel Example¶
import json
import os
from cicada.core.model import MultiModalModel
from cicada.core.utils import cprint
from dotenv import load_dotenv
from toolregistry.integrations.openapi import HttpClientConfig, load_openapi_spec
from toolregistry import ToolRegistry
load_dotenv()
# Initialize LLM model
model_name = os.getenv("MODEL", "deepseek-v3")
API_KEY = os.getenv("API_KEY", "your-api-key")
BASE_URL = os.getenv("BASE_URL", "https://api.deepseek.com/")
stream = os.getenv("STREAM", "True").lower() == "true"
llm = MultiModalModel(
api_key=API_KEY,
api_base_url=BASE_URL,
model_name=model_name,
stream=stream,
)
# Initialize tool registry
tool_registry = ToolRegistry()
client_config = HttpClientConfig(base_url="http://localhost:8000")
openapi_spec = load_openapi_spec("http://localhost:8000")
tool_registry.register_from_openapi(client_config, openapi_spec)
print(tool_registry.get_available_tools())
# Read input file
input_file = "examples/hub_related/concurrent_raw_results.txt"
with open(input_file) as f:
input_content = f.read()
instruction = f"""
I have a few test results from multiple runs. Please use the available tools to compute the averages of the metrics for each category.
The input is as {input_content}
"""
# Query LLM and fetch result
response = llm.query(instruction, tools=tool_registry, stream=stream)
cprint(json.dumps(response, indent=2))
OpenAI Client Example¶
import inspect
import os
from dotenv import load_dotenv
from toolregistry.integrations.openapi import HttpClientConfig, load_openapi_spec
from toolregistry import ToolRegistry
from openai import OpenAI
load_dotenv()
model_name = os.getenv("MODEL", "deepseek-v3")
stream = os.getenv("STREAM", "True").lower() == "true"
API_KEY = os.getenv("API_KEY", "your-api-key")
BASE_URL = os.getenv("BASE_URL", "https://api.deepseek.com/")
# Initialize tool registry
tool_registry = ToolRegistry()
client_config = HttpClientConfig(base_url="http://localhost:8000")
openapi_spec = load_openapi_spec("http://localhost:8000")
tool_registry.register_from_openapi(client_config, openapi_spec)
print(tool_registry.get_available_tools())
# Read input file
input_file = "examples/hub_related/concurrent_raw_results.txt"
with open(input_file) as f:
input_content = f.read()
# Set up OpenAI client
client = OpenAI(
api_key=os.getenv("API_KEY", "your-api-key"),
base_url=os.getenv("BASE_URL", "https://api.deepseek.com/"),
)
def handle_tool_calls(response, messages):
"""Handle tool calls in a loop until no more tool calls are needed"""
while response.choices[0].message.tool_calls:
tool_calls = response.choices[0].message.tool_calls
print("Tool calls:", tool_calls)
# Execute tool calls
tool_responses = tool_registry.execute_tool_calls(tool_calls)
# Construct assistant messages with results
assistant_tool_messages = tool_registry.build_tool_call_messages(
tool_calls, tool_responses
)
messages.extend(assistant_tool_messages)
# Send the results back to the model
response = client.chat.completions.create(
model=model_name,
messages=messages,
tools=tool_registry.get_schemas(),
tool_choice="auto",
)
return response
messages = [
{
"role": "user",
"content": inspect.cleandoc(f"""
I have a few test results from multiple runs. Please use the available tools to compute the averages of the metrics for each category.
The input is as {input_content}"""),
}
]
# Make the chat completion request
response = client.chat.completions.create(
model=model_name,
messages=messages,
tools=tool_registry.get_schemas(),
tool_choice="auto",
)
# Handle tool calls using the new function (without iteration limit)
response = handle_tool_calls(response, messages)
# Print final response
if response.choices[0].message.content:
print(response.choices[0].message.content)