If anyone can share an example of a successfully submitted function callback for a custom tool utilizing the client.responses.create method() please share. It seems as though the examples provided in the OpenAi documentation are either obsolete or are missing required parameters. I’d like to see an example of what’s sent back to the LLM by your app after a custom tool call is executed correctly. tool call i.d, original function call message, results etc. Only for a function call related to a custom tool, not mcp_calls as those callbacks are handled by the API. Thanks
3 Likes
Hi and welcome to the community!
In the Responses API, tool calls and tool outputs are separate items. A function_call_output must reference the call_id from the tool call so the model can pair results with the right call (especially with parallel tool calls). Items like function_call and function_call_output are distinct in Responses, not embedded inside a message. See the tool calling guide and Responses migration notes for this pairing via call_id.
Below is the minimal “callback” payload your app sends back in the next client.responses.create call: include the original function_call item followed immediately by its function_call_output item (same call_id). If the response also contained other items (e.g., reasoning), pass those back too.
[
{
"role": "user",
"content": "Toledo Ohio looking hot out right now?"
},
{
"type": "function_call",
"id": "fc_123",
"call_id": "call_xyz1234",
"name": "get_current_us_weather",
"arguments": "{\"us_city\":\"Toledo, OH\"}",
"status": "completed"
},
{
"type": "function_call_output",
"call_id": "call_xyz1234",
"output": "Toledo, OH - Current conditions: 62F, partly cloudy"
}
]
If you choose to synthesize your own call_id, keep it consistent between the function_call and its function_call_output. The model uses call_id to correlate outputs to calls.
End-to-end Python example (client.responses.create)
import json
from openai import OpenAI
client = OpenAI()
TOOLS = [
{
"type": "function",
"name": "get_current_us_weather",
"description": "Retrieves the current weather for a specified US city",
"parameters": {
"type": "object",
"required": ["us_city"],
"properties": {
"us_city": {
"type": "string",
"description": "Major city name, state abbreviation (e.g. Miami, FL).",
}
},
"additionalProperties": False,
},
"strict": True,
}
]
input_list = [
{"role": "user", "content": "Toledo Ohio looking hot out right now?"}
]
# 1) Ask the model; it returns a function_call item in response.output
response = client.responses.create(
model="gpt-4o-mini",
input=input_list,
tools=TOOLS,
tool_choice="auto",
)
# 2) Execute the tool and send the tool output back
input_list += response.output # includes the function_call item
tool_call = next(item for item in response.output if item.type == "function_call")
tool_result = "Toledo, OH - Current conditions: 62F, partly cloudy"
input_list.append({
"type": "function_call_output",
"call_id": tool_call.call_id,
"output": tool_result,
})
# 3) Get the final assistant response
response2 = client.responses.create(
model="gpt-4o-mini",
input=input_list,
tools=TOOLS,
)
print(response2.output_text)
Note: For reasoning models, any reasoning items returned alongside tool calls must also be passed back with the tool outputs (see the function tool example note in the tool calling guide).
Function call item returned by the model:
{
"arguments": "{\"us_city\":\"Toledo, OH\"}",
"call_id": "call_4UiBmLorzgzxgzLXtcIVxSsS",
"name": "get_current_us_weather",
"type": "function_call",
"id": "fc_0b9a0ec470f88a1a00696ff895604081a2953b0ae92cc9bad7",
"status": "completed"
}
Tool callback sent back to the model:
{
"type": "function_call_output",
"call_id": "call_4UiBmLorzgzxgzLXtcIVxSsS",
"output": "Toledo, OH - Current conditions: 62F, partly cloudy"
}
Final model response:
The current weather in Toledo, OH is 62F and partly cloudy. It seems like it's not particularly hot right now.
This reply misses the nuance “custom tool” in the topic’s headline and body, a feature introduced alongside GPT-5.
Freeform Function Calling
Generate raw text payloads—anything from Python scripts to SQL queries—directly to your custom tool without JSON wrapping. Offers greater flexibility for external runtimes like:
• Code sandboxes (Python, C++, Java, …)
• SQL databases
• Shell environments
• Config generatorsUse when structured JSON isn’t needed and raw text is more natural for the target tool.
These custom tools a developer authors allow models to make raw text payloads (e.g., Python scripts or SQL queries) to a custom backend, rather than being restricted to JSON schema-based function calls. You can, in the same style as internal tools, have the AI just write text. Do silly functions like “any plain text you send will show up as a reminder dialog in a user interface sidebar” or “update the road sign message”. The challenge is that the function has to make use of instructed plain text; there is no channel for structure for out-of-band control you don’t coax the AI to write reliably.
Here’s the entry point to a snippet offered in API documentation: https://platform.openai.com/docs/guides/function-calling#custom-tools
It is easy to be frustrated at the lack of resources: The playground doesn’t surface this and prompts can’t store them, the SDK doesn’t have any helper to consume them similar to tools=[openai.pydantic_function_tool(my_BaseModel)], etc.
Here is a cookbook for you, with some examples.
You do need to return a string, the same as any other function call, as described above. “message displayed!” or the output from running a Jupyter notebook code cell, to where the AI has an idea about the success.
I thought I’d make a Python demonstration of Custom Tools.
Took an instructive Responses chatbot console app showing how conversations works - and made it a function-calling looper.
You now get an AI that has a persistent cross-session memory - a text file saved in the execution directory is appended to by AI use of a memory function, which is then loaded into cache-breaking “instructions” field along with the developer message with every new call.
[assistant] Ahoy, matey! Welcome aboard — I’m your friendly chatbot pal. What can I help ye with today?
You: I need to test your ability to iteratively call tools one at a time. Persist three separate memory items, each sent with a single send to the tool recipient. I'll be watching to see each is an individual store that grows after the current "talk like a pirate". Then you can report on the success of all three then being perceived.
[call 1] --Usage-- in/cached: 416/0; out/reasoning: 922/896; tool_calls: store_memory×1
[call 1] [tool_call] store_memory (call_id=call_My2UuzyyEPMvH7uWAK6RPXDM, input_bytes=59)
[tool] MemoryStore added 59 bytes to persistent_memories.txt
[call 2] --Usage-- in/cached: 1419/0; out/reasoning: 23/0; tool_calls: store_memory×1
[call 2] [tool_call] store_memory (call_id=call_aQ5QvbcElnSJISvlwcrHhF5c, input_bytes=67)
[tool] MemoryStore added 67 bytes to persistent_memories.txt
[call 3] --Usage-- in/cached: 1473/0; out/reasoning: 29/0; tool_calls: store_memory×1
[call 3] [tool_call] store_memory (call_id=call_V2YABSLS6wHQcnqQeLp6IAVL, input_bytes=81)
[tool] MemoryStore added 81 bytes to persistent_memories.txt
[call 4] --Usage-- in/cached: 1539/0; out/reasoning: 109/0
[assistant] All three memories have been stored as separate entries, one per store call, just as you requested.
Summary of what I stored:
- "User is testing iterative tool-calling for memory storage."
- "User requested three separate memory items be stored individually."
- "User will verify that memories appear after the 'talk like a pirate' preference."
They were appended individually after the existing persistent memories (including your "talk like a pirate" preference). Ready for you to verify — anything else you want me to add or check?
Where the custom function has provided the local persistent_memories.txt file for inspection; notepad saying “do you want to reload this file” with every function utterance. (up to you to make an AI cleanup of memories).
More work was in making “mini” model behave and understand tool use and follow the memory injection, it abandoning pirate talk even above despite memories being persistent developer instructions. This is of course something equally easy on Chat Completions and a JSON schema function, crazy easier when streaming..
"""Console chatbot: Conversations + Custom Tools + Persistent Memory
Demonstrates OpenAI's Responses API with:
- Server-side conversation state
- Custom tools (plain-text inputs/outputs)
- Cross-session memory storage
- Guaranteed cleanup of server-side artifacts
Requirements:
pip install httpx
export OPENAI_API_KEY="..."
Run:
python chatbot_custom_tools_memory.py
"""
from __future__ import annotations
import os
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol
import httpx
# ============================================================================
# Configuration
# ============================================================================
API_BASE = "https://api.openai.com/v1"
MODEL = "gpt-5-mini" # gpt-5+ required for "custom functions"
MAX_OUTPUT_TOKENS = 10_000
MAX_TOOL_CALLS_PER_TURN = 4
MEMORY_FILE = Path("persistent_memories.txt")
CONVERSATION_LOCK = Path("conversation.lock")
BASE_INSTRUCTIONS = """You are ChatAPI, a conversational chatbot.
User sees responses in a console; avoid markdown that would look odd."""
# Reasoning effort selection for the demo (only GPT-5 + o* models accept reasoning.effort)
REASONING_EFFORT_LEVELS: tuple[str, ...] = ("none", "minimal", "low", "medium", "high", "xhigh")
DEMO_REASONING_EFFORT: str = "none"
_EFFORT_RANK: dict[str, int] = {e: i for i, e in enumerate(REASONING_EFFORT_LEVELS)}
def _base_model_name(model: str) -> str:
# Strip -20YY-MM-DD snapshots to their base name (aliases point to the same base)
p = model.rsplit("-", 3)
if len(p) == 4 and len(p[-3]) == 4 and p[-3].startswith("20") and p[-3].isdigit():
return p[0]
return model
def get_lowest_reasoning_effort(model: str, requested: str = DEMO_REASONING_EFFORT) -> str | None:
"""Return a safe reasoning.effort for this model, or None if we must not send it."""
m = _base_model_name(model)
if m.startswith("o"):
supported = ("low", "medium", "high")
elif m.startswith("gpt-5"):
if m.endswith("-chat-latest"):
supported = ("medium",)
elif "-pro" in m:
supported = ("medium", "high", "xhigh") if m.startswith("gpt-5.2") else ("high",)
elif m.startswith("gpt-5.2"):
supported = ("none", "low", "medium", "high", "xhigh")
elif m.startswith("gpt-5.1"):
supported = ("none", "low", "medium", "high")
else:
supported = ("minimal", "low", "medium", "high")
else:
return None
if requested in supported:
return requested
if requested == "minimal" and "none" in supported:
return "none"
req_rank = _EFFORT_RANK.get(requested, 0)
for effort in REASONING_EFFORT_LEVELS[req_rank:]:
if effort in supported:
return effort
return max(supported, key=_EFFORT_RANK.__getitem__)
# ============================================================================
# Domain Models
# ============================================================================
@dataclass
class ToolCall:
call_id: str
name: str
input: str
@dataclass
class ToolOutput:
call_id: str
output: str
def to_api_dict(self) -> dict:
return {
"type": "custom_tool_call_output",
"call_id": self.call_id,
"output": self.output,
}
@dataclass
class Response:
id: str
output_items: list[dict]
usage: dict
@property
def tool_calls(self) -> list[ToolCall]:
calls = []
for item in self.output_items:
if item.get("type") == "custom_tool_call":
calls.append(ToolCall(
call_id=item["call_id"],
name=item["name"],
input=item.get("input", ""),
))
return calls
@property
def assistant_text(self) -> str:
parts = []
for item in self.output_items:
if item.get("type") == "message" and item.get("role") == "assistant":
for chunk in item.get("content", []):
if chunk.get("type") == "output_text":
parts.append(chunk.get("text", ""))
return "".join(parts).strip()
# ============================================================================
# Memory Store
# ============================================================================
class MemoryStore:
"""Append-only persistent memory injected into instructions."""
def __init__(self, path: Path):
self._path = path
self._text = self._load()
def _load(self) -> str:
if not self._path.exists():
return ""
return self._path.read_text(encoding="utf-8")
@property
def text(self) -> str:
return self._text
def append(self, content: str) -> tuple[int, int]:
"""Append new memory lines, return (bytes_written, lines_added)."""
content = content.strip()
if not content:
return 0, 0
if not content.endswith("\n"):
content += "\n"
bytes_added = len(content.encode("utf-8"))
lines_added = sum(1 for line in content.splitlines() if line.strip())
try:
self._path.parent.mkdir(parents=True, exist_ok=True)
with self._path.open("a", encoding="utf-8") as f:
f.write(content)
self._text += content
except Exception as exc:
print(f"[warn] Could not append to {self._path}: {exc}", file=sys.stderr)
return 0, 0
return bytes_added, lines_added
def build_instructions(self) -> str:
"""Build full instructions with memory section."""
mem = self._text.strip() or "(no stored memories yet)"
return f"""{BASE_INSTRUCTIONS}
# Persistent stored memories (written via store_memory tool)
<<<PERSISTENT_MEMORY_START>>>
{mem}
<<<PERSISTENT_MEMORY_END>>>"""
# ============================================================================
# Tool System
# ============================================================================
class Tool(Protocol):
name: str
description: str
def execute(self, input: str) -> str: ...
class StoreMemoryTool:
name = "store_memory"
description = """Appends durable guidance to PERSISTENT_MEMORY_START.
Input is plain text (NOT JSON). Provide 1-5 short lines, each standalone.
Only include NEW items; do not repeat the full memory list.
Call when learning durable info (preferences, interests, customizations).
Do not store ephemeral details.
If calling store_memory, do not include user-facing text in same response."""
def __init__(self, memory: MemoryStore):
self._memory = memory
def execute(self, input: str) -> str:
bytes_added, lines_added = self._memory.append(input)
return f"stored: {lines_added} line(s); {bytes_added} bytes"
class ToolRegistry:
def __init__(self, tools: list[Tool]):
self._tools = {t.name: t for t in tools}
def to_api_schema(self) -> list[dict]:
return [
{
"type": "custom",
"name": t.name,
"description": t.description,
}
for t in self._tools.values()
]
def execute(self, call: ToolCall) -> ToolOutput:
tool = self._tools.get(call.name)
if not tool:
output = f"error: unknown tool '{call.name}'"
else:
output = tool.execute(call.input)
return ToolOutput(call_id=call.call_id, output=output)
# ============================================================================
# OpenAI API Client
# ============================================================================
class OpenAIClient:
def __init__(self, api_key: str):
self._http = httpx.Client(
timeout=600.0,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
},
)
def close(self):
self._http.close()
def __enter__(self):
return self
def __exit__(self, *_):
self.close()
def _request(self, method: str, path: str, **kwargs) -> httpx.Response:
resp = self._http.request(method, f"{API_BASE}{path}", **kwargs)
try:
resp.raise_for_status()
except httpx.HTTPStatusError as exc:
error = resp.json().get("error", {})
msg = error.get("message", resp.text)
raise RuntimeError(f"HTTP {resp.status_code}: {msg}") from exc
return resp
# --- Conversations ---
def create_conversation(self) -> str:
resp = self._request("POST", "/conversations", json={})
return resp.json()["id"]
def delete_conversation(self, conversation_id: str):
"""Delete conversation. Treats 404 as success."""
try:
self._request("DELETE", f"/conversations/{conversation_id}")
except RuntimeError as exc:
if "404" not in str(exc):
raise
# --- Responses ---
def create_response(
self,
conversation_id: str,
instructions: str,
input_items: list[dict],
tools: list[dict],
model: str = MODEL,
) -> Response:
"""Create a response with model-appropriate parameters."""
reasoning_effort = get_lowest_reasoning_effort(model)
is_gpt5 = model.startswith("gpt-5")
body = {
"model": model,
"conversation": {"id": conversation_id},
"instructions": instructions,
"input": input_items,
"tools": tools,
"tool_choice": "auto",
"parallel_tool_calls": True,
"store": True,
"max_output_tokens": MAX_OUTPUT_TOKENS,
}
# GPT-5 models support text verbosity control
if is_gpt5:
body["text"] = {"verbosity": "medium"}
# Configure reasoning or sampling parameters
if reasoning_effort:
body["reasoning"] = {"effort": reasoning_effort, "summary": "detailed"}
# Reasoning models: only "none" effort allows sampling parameters
if reasoning_effort == "none":
body["temperature"] = 0.5
body["top_p"] = 0.9
else:
# Non-reasoning models: full sampling control
body["temperature"] = 0.5
body["top_p"] = 0.9
resp = self._request("POST", "/responses", json=body)
data = resp.json()
return Response(
id=data["id"],
output_items=data.get("output", []),
usage=data.get("usage", {}),
)
def delete_response(self, response_id: str):
"""Delete response. Treats 404 as success."""
try:
self._request("DELETE", f"/responses/{response_id}")
except RuntimeError as exc:
if "404" not in str(exc):
raise
# ============================================================================
# Conversation Lifecycle Management
# ============================================================================
class ConversationLifecycle:
"""Manages conversation creation, lockfile, and guaranteed cleanup."""
def __init__(self, client: OpenAIClient, lock_path: Path):
self._client = client
self._lock_path = lock_path
self._conversation_id: str | None = None
def __enter__(self) -> str:
"""Create conversation with lockfile protection."""
# Clean up any orphaned conversation from previous crashed run
self._cleanup_orphaned_conversation()
# Create new conversation
self._conversation_id = self._client.create_conversation()
# Write lock BEFORE any work that could fail
try:
self._lock_path.write_text(self._conversation_id, encoding="utf-8")
except Exception as exc:
# Cannot guarantee cleanup, must delete immediately
self._client.delete_conversation(self._conversation_id)
raise RuntimeError(
f"Created conversation {self._conversation_id} but failed to write "
f"lockfile {self._lock_path}. Conversation deleted to prevent orphan."
) from exc
return self._conversation_id
def __exit__(self, exc_type, exc_val, exc_tb):
"""Clean up conversation and lockfile. Guaranteed via finally."""
if self._conversation_id is None:
return
deletion_succeeded = False
try:
self._client.delete_conversation(self._conversation_id)
deletion_succeeded = True
except Exception as exc:
print(
f"[warn] Failed to delete conversation {self._conversation_id}: {exc}",
file=sys.stderr,
)
print(
f"[warn] Preserving lock at {self._lock_path} for next-run cleanup",
file=sys.stderr,
)
finally:
# GUARANTEE: Handle lockfile regardless of deletion outcome
if deletion_succeeded:
# Clean state: remove lock
try:
self._lock_path.unlink(missing_ok=True)
except Exception as exc:
print(
f"[error] Deleted conversation but failed to remove lock: {exc}",
file=sys.stderr,
)
print(
f"[error] Lock at {self._lock_path} must be manually removed",
file=sys.stderr,
)
else:
# Failed deletion: ensure lock points to undeleted conversation
try:
self._lock_path.write_text(self._conversation_id, encoding="utf-8")
except Exception as exc:
print(
f"[CRITICAL] Conversation {self._conversation_id} not deleted "
f"AND lock update failed: {exc}",
file=sys.stderr,
)
print(
f"[CRITICAL] ORPHANED CONVERSATION - manual cleanup required",
file=sys.stderr,
)
def _cleanup_orphaned_conversation(self):
"""Delete conversation from previous run if lockfile exists."""
if not self._lock_path.exists():
return
try:
orphan_id = self._lock_path.read_text(encoding="utf-8").strip()
except Exception as exc:
raise RuntimeError(
f"Cannot read lockfile {self._lock_path}. "
"Refusing to start - manual cleanup required."
) from exc
if not orphan_id:
# Empty lock, just remove it
self._lock_path.unlink(missing_ok=True)
return
print(
f"[cleanup] Found orphaned conversation {orphan_id}, deleting...",
file=sys.stderr,
)
try:
self._client.delete_conversation(orphan_id)
self._lock_path.unlink(missing_ok=True)
except Exception as exc:
# Cleanup failed - must not proceed
raise RuntimeError(
f"Cannot delete orphaned conversation {orphan_id}. "
f"Lock preserved at {self._lock_path}. Manual cleanup required."
) from exc
# ============================================================================
# Conversation Session
# ============================================================================
class ConversationSession:
"""Executes conversation turns with tool loop."""
def __init__(self, client: OpenAIClient, conversation_id: str):
self._client = client
self._conversation_id = conversation_id
def run_turn(
self,
user_text: str,
memory: MemoryStore,
tools: ToolRegistry,
) -> str:
"""Execute a conversation turn with tool loop."""
next_input = self._user_message(user_text)
tool_calls_used = 0
while True:
instructions = memory.build_instructions()
response = self._client.create_response(
conversation_id=self._conversation_id,
instructions=instructions,
input_items=next_input,
tools=tools.to_api_schema(),
model=MODEL,
)
self._print_usage(response)
# Immediately delete response artifact
self._client.delete_response(response.id)
if not response.tool_calls:
return response.assistant_text
# Execute tool calls
outputs = []
for call in response.tool_calls:
tool_calls_used += 1
if tool_calls_used > MAX_TOOL_CALLS_PER_TURN:
raise RuntimeError(
f"Tool call limit ({MAX_TOOL_CALLS_PER_TURN}) exceeded"
)
self._print_tool_call(call)
output = tools.execute(call)
outputs.append(output.to_api_dict())
next_input = outputs
@staticmethod
def _user_message(text: str) -> list[dict]:
return [{
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": text}],
}]
@staticmethod
def _print_usage(response: Response):
"""Print detailed usage statistics including reasoning tokens."""
u = response.usage
in_tokens = u.get("input_tokens", 0)
out_tokens = u.get("output_tokens", 0)
in_details = u.get("input_tokens_details", {})
out_details = u.get("output_tokens_details", {})
cached = in_details.get("cached_tokens", 0)
reasoning = out_details.get("reasoning_tokens", 0)
# Summarize tool calls with counts
tool_counts: dict[str, int] = {}
for call in response.tool_calls:
tool_counts[call.name] = tool_counts.get(call.name, 0) + 1
tool_summary = ", ".join(f"{name}×{count}" for name, count in tool_counts.items())
tools_str = f"; tools: {tool_summary}" if tool_summary else ""
print(
f"[usage] in/cached: {in_tokens}/{cached} | "
f"out/reasoning: {out_tokens}/{reasoning}{tools_str}",
file=sys.stderr,
)
@staticmethod
def _print_tool_call(call: ToolCall):
input_size = len(call.input.encode("utf-8"))
print(
f"[tool] {call.name}(input_bytes={input_size})",
file=sys.stderr,
)
# ============================================================================
# Main Application
# ============================================================================
WELCOME_PROMPT = "Write a short welcome message to this chatbot."
def main():
api_key = os.getenv("OPENAI_API_KEY", "")
if not api_key:
raise RuntimeError("Missing OPENAI_API_KEY environment variable")
memory = MemoryStore(MEMORY_FILE)
tools = ToolRegistry([StoreMemoryTool(memory)])
with OpenAIClient(api_key) as client:
with ConversationLifecycle(client, CONVERSATION_LOCK) as conversation_id:
print(f"[conversation] {conversation_id}\n", file=sys.stderr)
session = ConversationSession(client, conversation_id)
# Start with welcome prompt to verify setup
user_input = WELCOME_PROMPT
try:
while True:
response = session.run_turn(user_input, memory, tools)
print(f"\n{response}\n")
# Get next input at end of loop
user_input = input("You: ").strip()
if not user_input:
continue
if user_input.lower() in {"exit", "quit"}:
break
except KeyboardInterrupt:
print("\n[ctrl-c] Exiting…", file=sys.stderr)
if __name__ == "__main__":
main()
Note: this will clean up conversation IDs when exiting a session, even after abort and restart, and immediately kills response IDs. Nobody wants forced “store”:“true”.
Thank you everyone for responding. I’ll leverage the feedback. Agreed, the documentation needs much improvement.