Documentation Index
Fetch the complete documentation index at: https://docs.danubeai.com/llms.txt
Use this file to discover all available pages before exploring further.
OpenAI Integration
This guide shows how to use Danube tools with OpenAI’s function calling feature, enabling GPT models to execute real-world actions through Danube’s tool marketplace.Overview
OpenAI’s function calling allows GPT models to:- Receive a list of available functions (tools)
- Decide which function to call based on user input
- Generate structured arguments for the function
- Process the function result and respond to the user
Prerequisites
pip install danube openai
Basic Example
Here’s a complete example of using Danube tools with OpenAI:import json
from openai import OpenAI
from danube import DanubeClient
# Initialize clients
openai_client = OpenAI(api_key="sk-...")
danube_client = DanubeClient(api_key="dk_...")
def danube_tool_to_openai_function(tool):
"""Convert a Danube tool to OpenAI function format."""
# Build properties from tool parameters
properties = {}
required = []
for name, param in tool.parameters.items():
param_type = param.get("type", "string")
# Map to JSON Schema types
json_type = {
"string": "string",
"integer": "integer",
"number": "number",
"boolean": "boolean",
"array": "array",
"object": "object",
}.get(param_type, "string")
properties[name] = {
"type": json_type,
"description": param.get("description", ""),
}
if param.get("required", False):
required.append(name)
return {
"type": "function",
"function": {
"name": tool.id, # Use tool ID for execution
"description": f"{tool.name}: {tool.description}",
"parameters": {
"type": "object",
"properties": properties,
"required": required,
},
},
}
def search_and_convert_tools(query: str, limit: int = 5):
"""Search Danube tools and convert to OpenAI format."""
tools = danube_client.tools.search(query, limit=limit)
return [danube_tool_to_openai_function(t) for t in tools]
def execute_tool_call(tool_call):
"""Execute an OpenAI tool call using Danube."""
tool_id = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
result = danube_client.tools.execute(
tool_id=tool_id,
parameters=arguments
)
return result.content if result.success else f"Error: {result.error}"
def chat_with_tools(user_message: str, tool_query: str = None):
"""Chat with GPT using Danube tools."""
# Get relevant tools
tools = search_and_convert_tools(tool_query or user_message)
# Initial API call
messages = [{"role": "user", "content": user_message}]
response = openai_client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=messages,
tools=tools,
tool_choice="auto",
)
assistant_message = response.choices[0].message
# Check if the model wants to call tools
if assistant_message.tool_calls:
messages.append(assistant_message)
# Execute each tool call
for tool_call in assistant_message.tool_calls:
print(f"Executing: {tool_call.function.name}")
result = execute_tool_call(tool_call)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
})
# Get final response
final_response = openai_client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=messages,
)
return final_response.choices[0].message.content
return assistant_message.content
# Example usage
if __name__ == "__main__":
response = chat_with_tools(
"What are the top stories on Hacker News right now?",
tool_query="hacker news"
)
print(response)
Full Agent Example
Here’s a more complete agent implementation with conversation history:import json
from typing import List, Dict, Any, Optional
from openai import OpenAI
from danube import DanubeClient
class DanubeOpenAIAgent:
"""An AI agent that uses Danube tools with OpenAI."""
def __init__(
self,
openai_api_key: str,
danube_api_key: str,
model: str = "gpt-4-turbo-preview",
system_prompt: str = None,
):
self.openai = OpenAI(api_key=openai_api_key)
self.danube = DanubeClient(api_key=danube_api_key)
self.model = model
self.messages: List[Dict[str, Any]] = []
self.available_tools: List[Dict] = []
self.tool_id_map: Dict[str, str] = {} # maps function name to tool_id
if system_prompt:
self.messages.append({"role": "system", "content": system_prompt})
def load_tools(self, query: str = "", service_id: str = None, limit: int = 20):
"""Load tools from Danube and convert to OpenAI format."""
if query:
tools = self.danube.tools.search(query, service_id=service_id, limit=limit)
elif service_id:
result = self.danube.services.get_tools(service_id, limit=limit)
tools = result.tools
else:
tools = self.danube.tools.search("", limit=limit)
self.available_tools = []
self.tool_id_map = {}
for tool in tools:
# Create safe function name (OpenAI requires alphanumeric + underscore)
safe_name = tool.name.replace(" ", "_").replace("-", "_")
safe_name = "".join(c for c in safe_name if c.isalnum() or c == "_")
self.tool_id_map[safe_name] = tool.id
# Build parameters
properties = {}
required = []
for name, param in tool.parameters.items():
param_info = param if isinstance(param, dict) else {}
properties[name] = {
"type": param_info.get("type", "string"),
"description": param_info.get("description", ""),
}
if param_info.get("required"):
required.append(name)
self.available_tools.append({
"type": "function",
"function": {
"name": safe_name,
"description": f"{tool.name}: {tool.description}"[:1024],
"parameters": {
"type": "object",
"properties": properties,
"required": required,
},
},
})
print(f"Loaded {len(self.available_tools)} tools")
return self
def _execute_tool(self, function_name: str, arguments: Dict) -> str:
"""Execute a tool and return the result."""
tool_id = self.tool_id_map.get(function_name)
if not tool_id:
return f"Error: Unknown tool '{function_name}'"
try:
result = self.danube.tools.execute(tool_id=tool_id, parameters=arguments)
return result.content if result.success else f"Error: {result.error}"
except Exception as e:
return f"Error executing tool: {str(e)}"
def chat(self, user_message: str, max_iterations: int = 5) -> str:
"""Send a message and get a response, executing tools as needed."""
self.messages.append({"role": "user", "content": user_message})
for _ in range(max_iterations):
# Call OpenAI
response = self.openai.chat.completions.create(
model=self.model,
messages=self.messages,
tools=self.available_tools if self.available_tools else None,
tool_choice="auto" if self.available_tools else None,
)
assistant_message = response.choices[0].message
# If no tool calls, we're done
if not assistant_message.tool_calls:
self.messages.append({
"role": "assistant",
"content": assistant_message.content
})
return assistant_message.content
# Add assistant message with tool calls
self.messages.append(assistant_message)
# Execute each tool call
for tool_call in assistant_message.tool_calls:
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
print(f" Tool: {function_name}")
print(f" Args: {arguments}")
result = self._execute_tool(function_name, arguments)
print(f" Result: {result[:200]}...")
self.messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
})
return "Max iterations reached"
def clear_history(self):
"""Clear conversation history (keeps system prompt)."""
self.messages = [m for m in self.messages if m.get("role") == "system"]
# Example usage
if __name__ == "__main__":
agent = DanubeOpenAIAgent(
openai_api_key="sk-...",
danube_api_key="dk_...",
system_prompt="You are a helpful assistant with access to various tools."
)
# Load tools for a specific domain
agent.load_tools(query="hacker news")
# Chat
response = agent.chat("What are the top 5 stories on Hacker News?")
print(response)
# Continue conversation
response = agent.chat("Tell me more about the first one")
print(response)
Dynamic Tool Loading
You can dynamically load tools based on the conversation:def dynamic_tool_agent(user_message: str):
"""Agent that dynamically loads relevant tools."""
# First, determine what tools might be needed
classification_response = openai_client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=[{
"role": "user",
"content": f"""Based on this user request, what type of tools would be helpful?
Return a JSON object with a 'tool_query' field containing search terms.
User request: {user_message}"""
}],
response_format={"type": "json_object"},
)
tool_info = json.loads(classification_response.choices[0].message.content)
tool_query = tool_info.get("tool_query", user_message)
# Load relevant tools
tools = search_and_convert_tools(tool_query, limit=10)
# Now handle the actual request with tools
return chat_with_tools(user_message, tools)
Best Practices
1. Tool Selection
Don’t load too many tools at once - it increases token usage and can confuse the model:# Good: Load specific tools
tools = client.tools.search("email", limit=5)
# Avoid: Loading everything
tools = client.tools.search("", limit=100) # Too many
2. Error Handling
Always handle tool execution errors gracefully:def safe_execute(tool_id: str, params: dict) -> str:
try:
result = danube_client.tools.execute(tool_id=tool_id, parameters=params)
if result.success:
return result.content
return f"Tool returned error: {result.error}"
except Exception as e:
return f"Failed to execute tool: {str(e)}"
3. Parameter Validation
Validate parameters before execution:def validate_and_execute(tool, arguments):
required_params = [p.name for p in tool.get_required_parameters()]
missing = [p for p in required_params if p not in arguments]
if missing:
return f"Missing required parameters: {missing}"
return danube_client.tools.execute(tool_id=tool.id, parameters=arguments)
4. Streaming Responses
For long-running tools, consider streaming:response = openai_client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=messages,
tools=tools,
stream=True,
)
for chunk in response:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="")
Complete Working Example
See the full example in the SDK repository:# Run the example
cd danube-python
python examples/openai_function_calling.py