7 Building an Agent with OpenAI’s Response API
7.1 Learning Objectives
- Understand how the Response API differs from the Chat Completion API
- Implement an agent using the Response API’s stateful conversation management
- Leverage the simplified tool handling in the Response API
- Compare the three approaches (SDK, Chat Completion, Response API) to agent development
In this chapter, we’ll implement our MarketMind agent using OpenAI’s Response API, which the OpenAI Agent SDK also leverages under the hood.
7.2 Understanding the Response API
The Response API represents OpenAI’s latest approach to building conversational AI systems. It introduces several improvements over the Chat Completion API. In particular, it has Built-in State Management: The API maintains conversation state through response IDs, eliminating the need to send the entire conversation history with each request.
Let’s create our Response API implementation:
touch src/agent_from_scratch/agent_response.pyWe are going to leverage the same tool management modules we built for the Chat Completion API implementation, and the tools we built for the Agent SDK implementation.
So let’s implement the agent:
# src/agent_from_scratch/agent_response.py
import json
import logging
import os
from openai import OpenAI
from src.agent_from_scratch.tool_manager import ToolManager
from src.common.config import DEFAULT_MODEL, SYSTEM_PROMPT, OPENAI_API_KEY, DEFAULT_MAX_ITERATIONS
# Configure logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG) # Set to DEBUG for detailed logs
class MarketMindResponseAgent:
"""
An AI-powered financial assistant using OpenAI's Response API with simplified state management.
"""
def __init__(self, model=DEFAULT_MODEL):
logger.debug(f"Initializing MarketMindResponseAgent with model: {model}")
self.tool_manager = ToolManager()
self.model = model
self.client = OpenAI(api_key=OPENAI_API_KEY)
# Initialize empty tool schemas
self.tool_schemas = []
# Track the most recent response ID for state management
self.previous_response_id = None
logger.debug(f"MarketMindResponseAgent initialized successfully with model: {model}")
def register_tool(self, name, description, tool_function):
"""Register a new tool with the agent."""
logger.debug(f"Registering tool: {name} - {description}")
self.tool_manager.register_tool(name, description, tool_function)
# Update tool schemas immediately - convert to Response API format
raw_schemas = self.tool_manager.get_schema_for_tools()
# Convert Chat Completions API format to Response API format
response_api_tools = []
for schema in raw_schemas:
if schema.get("type") == "function":
function_data = schema.get("function", {})
response_api_tools.append({
"type": "function",
"name": function_data.get("name", ""),
"description": function_data.get("description", ""),
"parameters": function_data.get("parameters", {})
})
self.tool_schemas = response_api_tools
logger.debug(f"Tool schemas updated: now have {len(self.tool_schemas)} tools")
return self
def _handle_function_call(self, function_call):
"""
Handle a function call from the Response API.
Args:
function_call: The function call object from the Response API
Returns:
The result of the function call as a string
"""
try:
function_name = function_call.name
function_args = json.loads(function_call.arguments)
logger.debug(f"Executing tool: {function_name} with args: {function_args}")
# Execute the tool
tool_result = self.tool_manager.execute_tool(function_name, **function_args)
logger.debug(f"Tool execution result: {tool_result}")
return str(tool_result)
except Exception as e:
error_msg = f"Error executing function call: {str(e)}"
logger.error(error_msg, exc_info=True)
return error_msg
def process_query(self, query):
"""
Process a user query using the Response API.
Args:
query: The user's query
Returns:
The agent's response as a string
"""
logger.debug(f"Processing query: {query}")
try:
# System prompt that defines the agent's capabilities
system_prompt = SYSTEM_PROMPT
# Prepare the initial input for the API call
input_messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": query}
]
logger.debug(f"Prepared initial input messages: {json.dumps(input_messages, indent=2)}")
logger.debug(f"Available tools: {json.dumps(self.tool_schemas, indent=2)}")
# Create the API call parameters
api_params = {
"model": self.model,
"input": input_messages,
"tools": self.tool_schemas
}
if self.previous_response_id:
api_params["previous_response_id"] = self.previous_response_id
logger.debug(f"Including previous response ID: {self.previous_response_id}")
# Call the API to get the initial response
logger.debug(f"Sending initial API request with parameters: {json.dumps(api_params, indent=2)}")
response = self.client.responses.create(**api_params)
logger.info(f"Received initial response with ID: {response.id}")
logger.debug(f"Initial response output: {json.dumps(response.model_dump(), indent=2)}")
# Save the response ID for future calls
self.previous_response_id = response.id
# Process the response and any subsequent responses with function calls
iteration = 0
# Continue processing responses until we get one without function calls
while iteration < DEFAULT_MAX_ITERATIONS:
iteration += 1
logger.debug(f"Starting iteration {iteration} of response processing loop")
# Process the current response
output = response.output
logger.debug(f"Processing {len(output)} output items in response {response.id}")
# Check if we have function calls to process
function_calls = [item for item in output if item.type == "function_call"]
if not function_calls:
logger.info(f"No function calls to process in iteration {iteration}")
break
logger.info(f"Found {len(function_calls)} function calls to process")
# Process all function calls in this response
function_call_results = []
for idx, function_call in enumerate(function_calls):
logger.debug(f"Processing function call {idx + 1}/{len(function_calls)}: {function_call.name}")
```python
# src/agent_from_scratch/agent_response.py (continued)
logger.info(f"Executing function call: {function_call.name} with arguments: {function_call.arguments}")
# Execute the function call
result = self._handle_function_call(function_call)
# Add the result to our list
function_call_results.append({
"type": "function_call_output",
"call_id": function_call.call_id,
"output": result
})
# Submit all function call results back to the API in one request
logger.debug(f"Sending follow-up request with {len(function_call_results)} function call results")
logger.debug(f"Function call results: {json.dumps(function_call_results, indent=2)}")
follow_up_response = self.client.responses.create(
model=self.model,
input=function_call_results,
previous_response_id=response.id,
tools=self.tool_schemas
)
logger.info(f"Received follow-up response with ID: {follow_up_response.id}")
logger.debug(f"Follow-up response output: {json.dumps(follow_up_response.model_dump(), indent=2)}")
# Update the response ID for future calls
self.previous_response_id = follow_up_response.id
response = follow_up_response
# Check if we hit the maximum number of iterations
if iteration >= DEFAULT_MAX_ITERATIONS:
logger.warning(f"Hit maximum iterations ({DEFAULT_MAX_ITERATIONS}) when processing function calls")
# Return the final text response
final_output = response.output_text
logger.info(f"Final response text: {final_output}")
return final_output
except Exception as e:
logger.error(f"Error processing query: {str(e)}", exc_info=True)
return f"An error occurred: {str(e)}"Let’s break down the changes:
7.2.1 Response ID-Based State Management
The most important aspect of the Response API related to conversation management is the Response ID:
# Track the most recent response ID for state management
self.previous_response_id = None
# Later in process_query:
if self.previous_response_id:
api_params["previous_response_id"] = self.previous_response_idThis simple mechanism represents a shift in how conversation state is managed:
Server-Side State: Instead of maintaining the entire conversation history on our side, OpenAI stores the conversation context associated with each response ID.
Token Efficiency: We no longer need to send the entire conversation history with each request, dramatically reducing token usage for long conversations.
Simplified Implementation: Our code only needs to track a single string (the response ID) rather than an ever-growing array of messages.
This approach is similar to how web sessions work - instead of sending all your user data with every request, the server stores the data and you just pass a session ID. It’s a much more scalable and efficient approach.
7.2.2 Tool Schema Transformation
The Response API also uses an evoled schema format than the Chat Completion API:
# Convert Chat Completions API format to Response API format
response_api_tools = []
for schema in raw_schemas:
if schema.get("type") == "function":
function_data = schema.get("function", {})
response_api_tools.append({
"type": "function",
"name": function_data.get("name", ""),
"description": function_data.get("description", ""),
"parameters": function_data.get("parameters", {})
})7.2.3 Batch Processing of Function Calls
Unlike the Chat Completion API where we handle tool calls one by one, here we process them in batch:
# Process all function calls in this response
function_call_results = []
for idx, function_call in enumerate(function_calls):
# Execute the function call
result = self._handle_function_call(function_call)
# Add the result to our list
function_call_results.append({
"type": "function_call_output",
"call_id": function_call.call_id,
"output": result
})
# Submit all function call results back to the API in one request
follow_up_response = self.client.responses.create(
model=self.model,
input=function_call_results,
previous_response_id=response.id,
tools=self.tool_schemas
)7.3 The Conversation Loop
The Response API’s conversation loop is similar to our Chat Completion implementation, but with some key differences:
# Continue processing responses until we get one without function calls
while iteration < DEFAULT_MAX_ITERATIONS:
# Process the current response
output = response.output
# Check if we have function calls to process
function_calls = [item for item in output if item.type == "function_call"]
if not function_calls:
break
# Process function calls and get a new response
# ...
# Update for next iteration
self.previous_response_id = follow_up_response.id
response = follow_up_responseThe key differences from our Chat Completion implementation:
Output Structure: Instead of checking
message.tool_calls, we examineresponse.outputfor items of type “function_call”.State Transition: We update
self.previous_response_idwith each new response, maintaining the conversation thread.No Message Array: We don’t need to maintain or update a growing message array - the Response API handles this for us.
This loop structure is cleaner and more focused on the essential flow: get response, check for function calls, execute them, submit results, repeat until done.
7.4 Update the CLI
Now, let’s add the Response API implementation to our CLI:
# Add this to src/cli/main.py after the chat_completion command
@cli.command()
@click.option('--model', default=DEFAULT_MODEL, help='The model to use for the agent')
@click.option('--debug', is_flag=True, help='Enable debug logging')
def response_api(model, debug):
"""Start MarketMind using the OpenAI Response API."""
# Set up logging - always log to file if debug is enabled, never to console for CLI
log_filename = setup_logging(
debug=debug,
module_loggers=DEFAULT_DEBUG_MODULES,
log_to_file=debug,
console_output=False # Don't output logs to console for CLI apps
)
logger.info(f"Starting MarketMind Response API Agent with model={model}")
# Initialize the agent
agent = MarketMindResponseAgent(model=model)
# Register all the tools
agent.register_tool(
"get_stock_price",
"Get the current price of a stock",
get_stock_price
)
agent.register_tool(
"get_stock_history",
"Get historical price data for a stock",
get_stock_history
)
agent.register_tool(
"get_company_info",
"Get basic information about a company",
get_company_info
)
agent.register_tool(
"get_financial_metrics",
"Get key financial metrics for a company",
get_financial_metrics
)
click.echo(click.style("\n🤖 MarketMind Financial Assistant powered by Response API", fg='blue', bold=True))
click.echo(click.style("Ask me about stocks, companies, or financial metrics. Type 'exit' to quit.\n", fg='blue'))
if log_filename:
click.echo(click.style(f"Log file: {log_filename}", fg='yellow'))
# Main conversation loop
while True:
# Get user input
user_input = click.prompt(click.style("You", fg='green', bold=True))
# Check for exit command
if user_input.lower() in ('exit', 'quit', 'q'):
logger.info("User requested exit")
click.echo(click.style("\nThank you for using MarketMind! Goodbye.", fg='blue'))
break
# Process the query
click.echo(click.style("MarketMind", fg='blue', bold=True) + " is thinking...")
click.echo(click.style(" 🤔 Processing query and deciding on actions...", fg="yellow"))
try:
# Process the query
response = agent.process_query(user_input)
click.echo(click.style(" ✅ Analysis complete, generating response...", fg="green"))
# Display the response
click.echo(click.style("MarketMind", fg='blue', bold=True) + f": {response}\n")
# Log response ID for debugging
if agent.previous_response_id:
logger.debug(f"Response ID captured for conversation continuity: {agent.previous_response_id}")
except Exception as e:
logger.error(f"Error processing query: {str(e)}", exc_info=True)
click.echo(click.style(" ❌ Error processing query", fg="red"))
click.echo(click.style("MarketMind", fg='blue', bold=True) +
f": I encountered an error while processing your request. Please try again.\n")Our CLI implementation for the Response API follows the same pattern as our previous implementations:
@cli.command()
@click.option('--model', default=DEFAULT_MODEL, help='The model to use for the agent')
@click.option('--debug', is_flag=True, help='Enable debug logging')
def response_api(model, debug):
"""Start MarketMind using the OpenAI Response API."""
# ...The CLI implementation also captures an important detail:
# Log response ID for debugging
if agent.previous_response_id:
logger.debug(f"Response ID captured for conversation continuity: {agent.previous_response_id}")This logging helps us verify that the response ID mechanism is working correctly, a crucial aspect of the Response API’s state management approach.
7.5 Testing the Response API Implementation
You can now run the agent with the Response API implementation:
market-mind response-apiThis provides a similar experience to the other implementations, but with more efficient state management.
7.6 Key Takeaways
In this chapter, we’ve: - Implemented an agent using the Response API - Leveraged response IDs for efficient state management - Used the batch processing approach for function calls - Added the Response API implementation to our CLI - Compared the three approaches to agent development
The Response API represents a step forward in making agent development more accessible and efficient. By understanding all three approaches—SDK, Chat Completion, and Response API—you now have a comprehensive toolkit for building AI agents that suit your specific needs.