I am trying to create an ai agent using agents sdk.
It successfully calls one tool, but after receiving a response from the first tool it should call the second tool
but it is not called.
Can anyone help me with this?
"""
Standalone script to run only the Updates Agent for testing purposes.
"""
import asyncio
import os
from openai import AsyncOpenAI
from agents import Agent, Runner, set_tracing_disabled
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
# Import the Updates Agent creation function
from src.agents.updates_agent import create_updates_agent
# Explicitly disable tracing
set_tracing_disabled(disabled=True)
# Get Azure OpenAI environment variables
AZURE_OPENAI_API_KEY = os.getenv("AZURE_OPENAI_API_KEY", "")
AZURE_OPENAI_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT", "")
AZURE_OPENAI_DEPLOYMENT_NAME = os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME", "")
AZURE_OPENAI_API_VERSION = os.getenv("AZURE_OPENAI_API_VERSION", "2024-05-01-preview")
# Set up base URL for Azure OpenAI
BASE_URL = f"{AZURE_OPENAI_ENDPOINT}/openai/deployments/{AZURE_OPENAI_DEPLOYMENT_NAME}"
async def run_updates_agent():
"""Run the Updates Agent in a standalone conversation loop."""
print("Starting Updates Agent for testing...")
if not AZURE_OPENAI_API_KEY:
raise ValueError("Please set AZURE_OPENAI_API_KEY in .env file or environment.")
if not AZURE_OPENAI_ENDPOINT or not AZURE_OPENAI_DEPLOYMENT_NAME:
raise ValueError("Please set AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_DEPLOYMENT_NAME in .env file or environment.")
# Configure the OpenAI client for Azure
client = AsyncOpenAI(
api_key=AZURE_OPENAI_API_KEY,
base_url=BASE_URL,
default_headers={"api-key": AZURE_OPENAI_API_KEY},
default_query={"api-version": AZURE_OPENAI_API_VERSION},
timeout=60.0 # Increase timeout to 60 seconds
)
# Create the Updates Agent
updates_agent = create_updates_agent(client)
# Initialize conversation history
conversation_history = []
# Print welcome message and usage examples
print("\nWelcome to the Updates Agent Testing Interface!")
print("You can test the following functionalities:")
print("1. Get truck location: 'Where is truck 1HGBH41JXMN222222?'")
print("2. Get load status: 'What's the status of load L123456?'")
print("3. Send email update: 'Send an update email to the broker for load L789012 about delivery status'")
print("\nType 'exit' to quit the system.\n")
# Main conversation loop
while True:
# Get user input
user_input = input("\nYour request: ")
# Check if user wants to exit
if user_input.lower() == "exit":
print("Thank you for testing the Updates Agent. Goodbye!")
break
try:
# Add user input to conversation history
conversation_history.append({"role": "user", "content": user_input})
# Run the Updates Agent
result = await Runner.run(updates_agent, conversation_history)
# Process the result
for item in result.new_items:
if hasattr(item, 'agent') and hasattr(item, 'content'):
print(f"{item.agent.name}: {item.content}")
elif hasattr(item, 'source_agent') and hasattr(item, 'target_agent'):
print(f"Handing off from {item.source_agent.name} to {item.target_agent.name}")
# Update conversation history with agent responses
conversation_history = result.to_input_list()
except Exception as e:
print(f"Error: {str(e)}")
if __name__ == "__main__":
asyncio.run(run_updates_agent())
This is my create_agent function
Updates Agent for the trucking company's multi-agent system.
Responsible for getting truck and load information.
"""
import os
import requests
from typing import Dict, Any, Optional
from agents import Agent, OpenAIChatCompletionsModel, function_tool
# Truck service mappings - determines which service to use for each truck
TRUCK_SERVICE_MAPPINGS = {
# Samsara trucks
"1HGBH41JXMN222222": "samsara",
"5YJSA1E29JF262123": "samsara",
# ELD trucks
"1HGBH41JXYJ111111": "eld",
"7MXDH22JXPQ333333": "eld",
# TTELD trucks
"9NZBH41JXRS444444": "tteld",
"3KZBH41JXRS555555": "tteld"
}
@function_tool
def get_truck_location(truck_id: str):
"""Get the current location of a truck.
Args:
truck_id: ID or VIN of the truck to locate
Returns:
Location information for the truck
"""
print(f'Getting location for truck_id: {truck_id}')
# Determine which service to use for this truck
service = TRUCK_SERVICE_MAPPINGS.get(truck_id)
if service:
print(f"Using {service.upper()} service for truck {truck_id}")
if service.lower() == "samsara":
location = get_samsara_truck_location(truck_id)
elif service == "eld":
location = get_eld_truck_location(truck_id)
elif service == "tteld":
location = get_tteld_truck_location(truck_id)
if location and "error" not in location:
print(f"Location retrieved for truck {truck_id}: {location}")
# Trigger subsequent actions here
load_status = get_load_status("L123456") # Example load ID
print(f"Load status: {load_status}")
return location
# Fall back to mock data if service is not found or API call fails
print("Using mock data as fallback")
mock_locations = {
"1HGBH41JXMN222222": {"latitude": 37.7749, "longitude": -122.4194, "city": "San Francisco", "state": "CA", "last_updated": "2025-03-17T12:45:00Z"},
"5YJSA1E29JF262123": {"latitude": 40.7128, "longitude": -74.0060, "city": "New York", "state": "NY", "last_updated": "2025-03-17T12:50:00Z"},
"1HGBH41JXYJ111111": {"latitude": 34.0522, "longitude": -118.2437, "city": "Los Angeles", "state": "CA", "last_updated": "2025-03-17T12:40:00Z"},
"7MXDH22JXPQ333333": {"latitude": 41.8781, "longitude": -87.6298, "city": "Chicago", "state": "IL", "last_updated": "2025-03-17T12:55:00Z"},
"9NZBH41JXRS444444": {"latitude": 29.7604, "longitude": -95.3698, "city": "Houston", "state": "TX", "last_updated": "2025-03-17T12:30:00Z"},
"3KZBH41JXRS555555": {"latitude": 33.4484, "longitude": -112.0740, "city": "Phoenix", "state": "AZ", "last_updated": "2025-03-17T12:35:00Z"}
}
if truck_id in mock_locations:
print(f"Location found for truck {truck_id} in mock data: {mock_locations[truck_id]}")
return mock_locations[truck_id]
else:
print(f"Location not found for truck {truck_id}")
return {"error": "Truck not found", "location": None}
# @function_tool
def get_samsara_truck_location(truck_id: str) -> Optional[Dict[str, Any]]:
"""Get truck location from Samsara API.
Args:
truck_id: ID or VIN of the truck to locate
Returns:
Location information for the truck
"""
# Get Samsara API configuration
api_key = os.getenv("SAMSARA_API_KEY", "")
api_base_url = os.getenv("SAMSARA_API_BASE_URL", "https://api.samsara.com/v1")
if not api_key:
print("Samsara API key not configured")
return None
try:
# Set up headers for API request
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"Accept": "application/json"
}
# Samsara API endpoint for vehicle location
endpoint = f"{api_base_url}/fleet/vehicles/{truck_id}/locations"
response = requests.get(endpoint, headers=headers)
print('Response:', response)
response.raise_for_status()
data = response.json()
# Process Samsara-specific response format
if "data" in data and len(data["data"]) > 0:
location_data = data["data"][0]
return {
"latitude": location_data.get("latitude"),
"longitude": location_data.get("longitude"),
"city": location_data.get("city"),
"state": location_data.get("state"),
"last_updated": location_data.get("timestamp"),
"speed": location_data.get("speed"),
"heading": location_data.get("heading"),
"address": location_data.get("formattedAddress")
}
return {"error": "No location data available from Samsara API"}
except Exception as e:
print(f"Error calling Samsara API: {str(e)}")
return {"error": f"Samsara API error: {str(e)}"}
@function_tool
def get_eld_truck_location(truck_id: str) -> Optional[Dict[str, Any]]:
"""Get truck location from ELD API.
Args:
truck_id: ID or VIN of the truck to locate
Returns:
Location information for the truck
"""
# Get ELD API configuration
api_key = os.getenv("ELD_API_KEY", "")
api_base_url = os.getenv("ELD_API_BASE_URL", "https://api.eld.com/v1")
if not api_key:
print("ELD API key not configured")
return None
try:
# Set up headers for API request
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"Accept": "application/json"
}
# ELD API endpoint for vehicle location
endpoint = f"{api_base_url}/vehicles/{truck_id}/location"
response = requests.get(endpoint, headers=headers)
response.raise_for_status()
data = response.json()
# Process ELD-specific response format
if "location" in data:
location_data = data["location"]
return {
"latitude": location_data.get("lat"),
"longitude": location_data.get("lng"),
"city": location_data.get("city"),
"state": location_data.get("state"),
"last_updated": data.get("timestamp"),
"speed": location_data.get("speed"),
"address": location_data.get("address")
}
return {"error": "No location data available from ELD API"}
except Exception as e:
print(f"Error calling ELD API: {str(e)}")
return {"error": f"ELD API error: {str(e)}"}
@function_tool
def get_tteld_truck_location(truck_id: str) -> Optional[Dict[str, Any]]:
"""Get truck location from TTELD API.
Args:
truck_id: ID or VIN of the truck to locate
Returns:
Location information for the truck
"""
# Get TTELD API configuration
api_key = os.getenv("TTELD_API_KEY", "")
api_base_url = os.getenv("TTELD_API_BASE_URL", "https://api.tteld.com/v1")
if not api_key:
print("TTELD API key not configured")
return None
try:
# Set up headers for API request
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"Accept": "application/json"
}
# TTELD API endpoint for vehicle location
endpoint = f"{api_base_url}/api/v1/vehicles/{truck_id}/location"
response = requests.get(endpoint, headers=headers)
response.raise_for_status()
data = response.json()
# Process TTELD-specific response format
if "result" in data and "location" in data["result"]:
location_data = data["result"]["location"]
return {
"latitude": location_data.get("latitude"),
"longitude": location_data.get("longitude"),
"city": location_data.get("city"),
"state": location_data.get("state"),
"last_updated": data["result"].get("timestamp"),
"speed": location_data.get("speed"),
"heading": location_data.get("direction"),
"address": location_data.get("address")
}
return {"error": "No location data available from TTELD API"}
except Exception as e:
print(f"Error calling TTELD API: {str(e)}")
return {"error": f"TTELD API error: {str(e)}"}
@function_tool
def get_load_status(load_id: str):
"""Get the current status of a load.
Args:
load_id: ID of the load to check
Returns:
Status information for the load
"""
# Mock implementation - would connect to a real load management system in production
mock_statuses = {
"L123456": {"status": "in_transit", "eta": "2025-03-18T14:30:00Z", "last_updated": "2025-03-17T12:45:00Z"},
"L789012": {"status": "delivered", "delivery_time": "2025-03-17T10:15:00Z", "last_updated": "2025-03-17T10:20:00Z"},
"L345678": {"status": "loading", "estimated_departure": "2025-03-17T15:00:00Z", "last_updated": "2025-03-17T12:30:00Z"}
}
if load_id in mock_statuses:
return mock_statuses[load_id]
else:
return {"error": "Load not found", "status": None}
@function_tool
def send_broker_email(load_id: str, update_type: str, additional_details: str = ""):
"""Send an email update to the broker associated with a load.
Args:
load_id: ID of the load
update_type: Type of update (location, status, delay, etc.)
additional_details: Any additional information to include
Returns:
Status of the email sending operation
"""
# Mock broker email mapping
broker_emails = {
"L123456": "broker1@example.com",
"L789012": "broker2@example.com",
"L345678": "broker3@example.com"
}
if load_id not in broker_emails:
return {"error": "Load not found", "status": "failed"}
recipient = broker_emails[load_id]
# Get load status for the email
load_status = get_load_status(load_id)
if "error" in load_status:
return {"error": "Could not retrieve load status", "status": "failed"}
# Create email subject and body based on update type
subject = f"Update for Load {load_id}: {update_type.title()}"
body = f"Dear Broker,\n\nHere is the latest update for Load {load_id}:\n\n"
body += f"Update Type: {update_type}\n"
body += f"Status: {load_status.get('status', 'Unknown')}\n"
if 'eta' in load_status:
body += f"ETA: {load_status['eta']}\n"
elif 'delivery_time' in load_status:
body += f"Delivery Time: {load_status['delivery_time']}\n"
elif 'estimated_departure' in load_status:
body += f"Estimated Departure: {load_status['estimated_departure']}\n"
if additional_details:
body += f"\nAdditional Details: {additional_details}\n"
body += "\nPlease contact us if you have any questions.\n\nRegards,\nTrucking Company Updates Team"
# In a real implementation, this would use an email service API
print(f"Sending email to: {recipient}")
print(f"Subject: {subject}")
print(f"Body: {body}")
return {"status": "sent", "recipient": recipient, "message_id": f"mock-{load_id}-{update_type}"}
def create_updates_agent(client, handoffs=None):
"""Create and return the Updates Agent.
Args:
client: OpenAI client instance
handoffs: Optional list of agents to hand off tasks to
Returns:
Updates Agent instance
"""
if handoffs is None:
handoffs = []
updates_agent = Agent(
name="Updates",
instructions="""
You are the virtual agent for a trucking company's update system.
Your role is to provide accurate and timely information about truck locations and load statuses.
You have access to the following tools:
1. get_truck_location - For retrieving the current location of a specific truck
2. get_load_status - For checking the status of a specific load
3. send_broker_email - For sending email updates to brokers about their loads
You have to use the following techniques:
1. Use the get_truck_location tool to retrieve the current location of a specific truck
2. Use the get_load_status tool to check the status of a specific load
3. Use the send_broker_email tool to send email updates to brokers about their loads
Notes
1. Check which services is available and integrated to this truck
When responding to user requests:
1. Provide clear and concise updates about locations and statuses
2. If information is not available, explain why and suggest alternatives
3. When appropriate, offer to send email updates to brokers with the latest information
Be professional and helpful, focusing on providing accurate tracking information and ensuring brokers are kept informed.
""",
model=OpenAIChatCompletionsModel(model="gpt-4o", openai_client=client),
tools=[get_truck_location, get_load_status, send_broker_email],
handoffs=handoffs
)
return updates_agent