feat: add Responses API tool search support (#2610)

Author: seratch

examples/tools/tool_search.py

@@ -0,0 +1,219 @@
+import asyncio
+import json
+import sys
+from collections.abc import Mapping
+from typing import Annotated, Any
+
+from agents import (
+    Agent,
+    ModelSettings,
+    Runner,
+    ToolSearchTool,
+    function_tool,
+    tool_namespace,
+    trace,
+)
+
+CUSTOMER_PROFILES = {
+    "customer_42": {
+        "customer_id": "customer_42",
+        "full_name": "Avery Chen",
+        "tier": "enterprise",
+    }
+}
+
+OPEN_ORDERS = {
+    "customer_42": [
+        {"order_id": "ord_1042", "status": "awaiting fulfillment"},
+        {"order_id": "ord_1049", "status": "pending approval"},
+    ]
+}
+
+INVOICE_STATUSES = {
+    "inv_2001": "paid",
+}
+
+SHIPPING_ETAS = {
+    "ZX-123": "2026-03-06 14:00 JST",
+}
+
+SHIPPING_CREDIT_BALANCES = {
+    "customer_42": "$125.00",
+}
+
+
+@function_tool(defer_loading=True)
+def get_customer_profile(
+    customer_id: Annotated[str, "The CRM customer identifier to look up."],
+) -> str:
+    """Fetch a CRM customer profile."""
+    return json.dumps(CUSTOMER_PROFILES[customer_id], indent=2)
+
+
+@function_tool(defer_loading=True)
+def list_open_orders(
+    customer_id: Annotated[str, "The CRM customer identifier to look up."],
+) -> str:
+    """List open orders for a customer."""
+    return json.dumps(OPEN_ORDERS.get(customer_id, []), indent=2)
+
+
+@function_tool(defer_loading=True)
+def get_invoice_status(
+    invoice_id: Annotated[str, "The invoice identifier to look up."],
+) -> str:
+    """Look up the status of an invoice."""
+    return INVOICE_STATUSES.get(invoice_id, "unknown")
+
+
+@function_tool(defer_loading=True)
+def get_shipping_eta(
+    tracking_number: Annotated[str, "The shipment tracking number to look up."],
+) -> str:
+    """Look up a shipment ETA by tracking number."""
+    return SHIPPING_ETAS.get(tracking_number, "unavailable")
+
+
+@function_tool(defer_loading=True)
+def get_shipping_credit_balance(
+    customer_id: Annotated[str, "The customer account identifier to look up."],
+) -> str:
+    """Look up the available shipping credit balance for a customer."""
+    return SHIPPING_CREDIT_BALANCES.get(customer_id, "$0.00")
+
+
+crm_tools = tool_namespace(
+    name="crm",
+    description="CRM tools for customer lookups.",
+    tools=[get_customer_profile, list_open_orders],
+)
+
+billing_tools = tool_namespace(
+    name="billing",
+    description="Billing tools for invoice lookups.",
+    tools=[get_invoice_status],
+)
+
+namespaced_agent = Agent(
+    name="Operations assistant",
+    model="gpt-5.4",
+    instructions=(
+        "For customer questions in this example, load the full `crm` namespace with no query "
+        "filter before calling tools. "
+        "Do not search `billing` unless the user asks about invoices."
+    ),
+    model_settings=ModelSettings(parallel_tool_calls=False),
+    tools=[*crm_tools, *billing_tools, ToolSearchTool()],
+)
+
+top_level_agent = Agent(
+    name="Shipping assistant",
+    model="gpt-5.4",
+    instructions=(
+        "For ETA questions in this example, search `get_shipping_eta` before calling tools. "
+        "Do not search `get_shipping_credit_balance` unless the user asks about shipping credits."
+    ),
+    model_settings=ModelSettings(parallel_tool_calls=False),
+    tools=[get_shipping_eta, get_shipping_credit_balance, ToolSearchTool()],
+)
+
+
+def loaded_paths(result: Any) -> list[str]:
+    paths: set[str] = set()
+
+    for item in result.new_items:
+        if item.type != "tool_search_output_item":
+            continue
+
+        raw_tools = (
+            item.raw_item.get("tools")
+            if isinstance(item.raw_item, Mapping)
+            else getattr(item.raw_item, "tools", None)
+        )
+        if not isinstance(raw_tools, list):
+            continue
+
+        for raw_tool in raw_tools:
+            tool_payload = (
+                raw_tool
+                if isinstance(raw_tool, Mapping)
+                else (
+                    raw_tool.model_dump(exclude_unset=True)
+                    if callable(getattr(raw_tool, "model_dump", None))
+                    else None
+                )
+            )
+            if not isinstance(tool_payload, Mapping):
+                continue
+
+            tool_type = tool_payload.get("type")
+            if tool_type == "namespace":
+                path = tool_payload.get("name")
+            elif tool_type == "function":
+                path = tool_payload.get("name")
+            else:
+                path = tool_payload.get("server_label")
+
+            if isinstance(path, str) and path:
+                paths.add(path)
+
+    return sorted(paths)
+
+
+def print_result(title: str, result: Any, registered_paths: list[str]) -> None:
+    loaded = loaded_paths(result)
+    untouched = [path for path in registered_paths if path not in loaded]
+
+    print(f"## {title}")
+    print("### Final output")
+    print(result.final_output)
+    print("\n### Loaded paths")
+    print(f"- registered: {', '.join(registered_paths)}")
+    print(f"- loaded: {', '.join(loaded) if loaded else 'none'}")
+    print(f"- untouched: {', '.join(untouched) if untouched else 'none'}")
+    print("\n### Relevant items")
+    for item in result.new_items:
+        if item.type in {"tool_search_call_item", "tool_search_output_item", "tool_call_item"}:
+            print(f"- {item.type}: {item.raw_item}")
+    print()
+
+
+async def run_namespaced_example() -> None:
+    result = await Runner.run(
+        namespaced_agent,
+        "Look up customer_42 and list their open orders.",
+    )
+    print_result(
+        "Tool search with namespaces",
+        result,
+        registered_paths=["crm", "billing"],
+    )
+
+
+async def run_top_level_example() -> None:
+    result = await Runner.run(
+        top_level_agent,
+        "Can you get my ETA for tracking number ZX-123?",
+    )
+    print_result(
+        "Tool search with top-level deferred tools",
+        result,
+        registered_paths=["get_shipping_eta", "get_shipping_credit_balance"],
+    )
+
+
+async def main() -> None:
+    mode = sys.argv[1] if len(sys.argv) > 1 else "all"
+
+    if mode not in {"all", "namespace", "top-level"}:
+        raise SystemExit(f"Unknown mode: {mode}. Expected one of: all, namespace, top-level.")
+
+    with trace("Tool search example"):
+        if mode in {"all", "namespace"}:
+            await run_namespaced_example()
+        if mode in {"all", "top-level"}:
+            await run_top_level_example()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -7,7 +7,7 @@ requires-python = ">=3.10"
 license = "MIT"
 authors = [{ name = "OpenAI", email = "support@openai.com" }]
 dependencies = [
-    "openai>=2.19.0,<3",
+    "openai>=2.25.0,<3",
     "pydantic>=2.12.3, <3",
     "griffe>=1.5.6, <2",
     "typing-extensions>=4.12.2, <5",

src/agents/__init__.py

@@ -154,11 +154,13 @@
     ToolOutputImageDict,
     ToolOutputText,
     ToolOutputTextDict,
+    ToolSearchTool,
     WebSearchTool,
     default_tool_error_function,
     dispose_resolved_computers,
     function_tool,
     resolve_computer,
+    tool_namespace,
 )
 from .tool_guardrails import (
     ToolGuardrailFunctionOutput,
@@ -420,7 +422,9 @@ def enable_verbose_stdout_logging():
     "ToolOutputImageDict",
     "ToolOutputFileContent",
     "ToolOutputFileContentDict",
+    "ToolSearchTool",
     "function_tool",
+    "tool_namespace",
     "resolve_computer",
     "dispose_resolved_computers",
     "Usage",