feat: add Programmatic Tool Calling support

examples/tools/programmatic_tool_calling.py

@@ -0,0 +1,129 @@
+import asyncio
+from typing import Literal
+
+from openai.types.responses import ResponseFunctionToolCall
+from openai.types.responses.response_output_item import Program
+from pydantic import BaseModel
+
+from agents import (
+    Agent,
+    ModelSettings,
+    ProgrammaticToolCallingTool,
+    Runner,
+    ToolCallItem,
+    function_tool,
+)
+
+Sku = Literal["desk-lamp", "ergonomic-keyboard", "usb-c-dock"]
+
+inventory: dict[Sku, int] = {
+    "desk-lamp": 12,
+    "ergonomic-keyboard": 7,
+    "usb-c-dock": 22,
+}
+
+weekly_demand: dict[Sku, int] = {
+    "desk-lamp": 18,
+    "ergonomic-keyboard": 16,
+    "usb-c-dock": 14,
+}
+
+inbound_units: dict[Sku, int] = {
+    "desk-lamp": 4,
+    "ergonomic-keyboard": 2,
+    "usb-c-dock": 0,
+}
+
+
+class InventoryOutput(BaseModel):
+    sku: Sku
+    available_units: int
+
+
+class WeeklyDemandOutput(BaseModel):
+    sku: Sku
+    forecast_units: int
+
+
+class InboundUnitsOutput(BaseModel):
+    sku: Sku
+    inbound_units: int
+
+
+@function_tool(allowed_callers=["programmatic"])
+def get_inventory(sku: Sku) -> InventoryOutput:
+    """Return the currently available units for one SKU."""
+    print(f"[tool] get_inventory({sku})")
+    return InventoryOutput(sku=sku, available_units=inventory[sku])
+
+
+@function_tool(allowed_callers=["programmatic"])
+def get_weekly_demand(sku: Sku) -> WeeklyDemandOutput:
+    """Return forecast demand for one SKU for the next seven days."""
+    print(f"[tool] get_weekly_demand({sku})")
+    return WeeklyDemandOutput(sku=sku, forecast_units=weekly_demand[sku])
+
+
+@function_tool(allowed_callers=["programmatic"])
+def get_inbound_units(sku: Sku) -> InboundUnitsOutput:
+    """Return units already scheduled to arrive for one SKU."""
+    print(f"[tool] get_inbound_units({sku})")
+    return InboundUnitsOutput(sku=sku, inbound_units=inbound_units[sku])
+
+
+async def main() -> None:
+    agent = Agent(
+        name="Replenishment planner",
+        model="gpt-5.6",
+        instructions="""
+<tool_orchestration>
+Use Programmatic Tool Calling to prepare a replenishment plan for desk-lamp,
+ergonomic-keyboard, and usb-c-dock. For every SKU, call get_inventory,
+get_weekly_demand, and get_inbound_units. Create all nine tool-call promises
+before awaiting them, then run them concurrently with one Promise.all call.
+
+Use a safety stock of 5 units. Calculate reorder_units as
+max(forecast_units + 5 - available_units - inbound_units, 0). In the program,
+return exactly one JSON object with recommendations and total_reorder_units.
+Each recommendation must include sku, available_units, forecast_units,
+inbound_units, and reorder_units. Include only positive reorder quantities and
+sort recommendations by reorder_units descending.
+
+Do not call these tools directly. In the final answer, explain the plan using
+the source values returned by the program.
+</tool_orchestration>
+        """.strip(),
+        model_settings=ModelSettings(tool_choice="programmatic_tool_calling"),
+        tools=[
+            get_inventory,
+            get_weekly_demand,
+            get_inbound_units,
+            ProgrammaticToolCallingTool(),
+        ],
+    )
+
+    result = await Runner.run(
+        agent,
+        "Which products should we reorder this week, and in what quantities?",
+    )
+
+    programmatic_calls: list[str] = []
+    for item in result.new_items:
+        if not isinstance(item, ToolCallItem):
+            continue
+        raw_item = item.raw_item
+        if isinstance(raw_item, Program):
+            print(f"\nGenerated program:\n{raw_item.code}\n")
+        elif (
+            isinstance(raw_item, ResponseFunctionToolCall)
+            and raw_item.caller is not None
+            and raw_item.caller.type == "program"
+        ):
+            programmatic_calls.append(raw_item.name)
+
+    print(f"Programmatic calls: {', '.join(programmatic_calls)}")
+    print(f"\nFinal answer:\n{result.final_output}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/agents/__init__.py

@@ -156,6 +156,7 @@
     MCPToolApprovalFunction,
     MCPToolApprovalFunctionResult,
     MCPToolApprovalRequest,
+    ProgrammaticToolCallingTool,
     ShellActionRequest,
     ShellCallData,
     ShellCallOutcome,
@@ -179,6 +180,7 @@
     ShellToolLocalSkill,
     ShellToolSkillReference,
     Tool,
+    ToolCaller,
     ToolOrigin,
     ToolOriginType,
     ToolOutputFileContent,
@@ -510,7 +512,9 @@ def enable_verbose_stdout_logging():
     "ApplyPatchTool",
     "ApplyPatchToolCustomDataContext",
     "ApplyPatchToolCustomDataExtractor",
+    "ProgrammaticToolCallingTool",
     "Tool",
+    "ToolCaller",
     "WebSearchTool",
     "HostedMCPTool",
     "MCPToolApprovalFunction",

src/agents/function_schema.py

@@ -40,6 +40,8 @@ class FuncSchema:
     strict_json_schema: bool = True
     """Whether the JSON schema is in strict mode. We **strongly** recommend setting this to True,
     as it increases the likelihood of correct JSON input."""
+    return_annotation: Any = inspect.Signature.empty
+    """The resolved return annotation, including `Annotated` metadata when present."""
 
     def to_call_args(self, data: BaseModel) -> tuple[list[Any], dict[str, Any]]:
         """
@@ -474,4 +476,5 @@ def function_schema(
         signature=sig,
         takes_context=takes_context,
         strict_json_schema=strict_json_schema,
+        return_annotation=type_hints_with_extras.get("return", sig.return_annotation),
     )

src/agents/items.py

@@ -45,13 +45,15 @@
     McpApprovalRequest,
     McpCall,
     McpListTools,
+    Program,
+    ProgramOutput,
 )
 from openai.types.responses.response_reasoning_item import ResponseReasoningItem
 from pydantic import BaseModel
 from typing_extensions import assert_never
 
 from ._tool_identity import FunctionToolLookupKey, get_function_tool_lookup_key, tool_trace_name
-from .exceptions import AgentsException, ModelBehaviorError
+from .exceptions import AgentsException, ModelBehaviorError, UserError
 from .logger import logger
 from .tool import (
     ToolOrigin,
@@ -60,6 +62,7 @@
     ToolOutputText,
     ValidToolOutputPydanticModels,
     ValidToolOutputPydanticModelsTypeAdapter,
+    _is_programmatic_tool_call,
 )
 from .usage import Usage
 from .util._json import _to_dump_compatible
@@ -85,6 +88,7 @@
 
 # Distinguish a missing dict entry from an explicit None value.
 _MISSING_ATTR_SENTINEL = object()
+_JSON_OUTPUT_ADAPTER = pydantic.TypeAdapter(Any)
 
 
 @dataclass
@@ -339,6 +343,7 @@ def release_agent(self) -> None:
     | ResponseCodeInterpreterToolCall
     | ImageGenerationCall
     | LocalShellCall
+    | Program
     | dict[str, Any]
 )
 """A type that represents a tool call item."""
@@ -382,6 +387,7 @@ def call_id(self) -> str | None:
     | ComputerCallOutput
     | LocalShellCallOutput
     | ResponseFunctionShellToolCallOutput
+    | ProgramOutput
     | dict[str, Any]
 )
 
@@ -785,7 +791,12 @@ def text_message_output(cls, message: MessageOutputItem) -> str:
 
     @classmethod
     def tool_call_output_item(
-        cls, tool_call: ResponseFunctionToolCall, output: Any
+        cls,
+        tool_call: ResponseFunctionToolCall,
+        output: Any,
+        *,
+        output_json_schema: dict[str, Any] | None = None,
+        output_type_adapter: pydantic.TypeAdapter[Any] | None = None,
     ) -> FunctionCallOutput:
         """Creates a tool call output item from a tool call and its output.
 
@@ -794,20 +805,108 @@ def tool_call_output_item(
         provided as Pydantic models or dicts, or an iterable of such items.
         """
 
-        converted_output = cls._convert_tool_output(output)
+        converted_output: str | ResponseFunctionCallOutputItemListParam
+        if output_type_adapter is not None:
+            try:
+                validated_output = (
+                    output_type_adapter.validate_json(output)
+                    if isinstance(output, str)
+                    else output_type_adapter.validate_python(output)
+                )
+            except pydantic.ValidationError as error:
+                raise UserError(
+                    "Function tool output does not match its declared output schema."
+                ) from error
+            dumped_output = output_type_adapter.dump_python(
+                validated_output,
+                mode="json",
+                by_alias=True,
+            )
+            if not isinstance(dumped_output, Mapping):
+                raise UserError("Function tool output schema requires a JSON object.")
+            converted_output = json.dumps(
+                dict(dumped_output),
+                ensure_ascii=False,
+                separators=(",", ":"),
+            )
+        elif output_json_schema is not None:
+            if isinstance(output, str):
+                try:
+                    dumped_output = json.loads(output)
+                except json.JSONDecodeError as error:
+                    raise UserError(
+                        "Function tool output schema requires a JSON object."
+                    ) from error
+            else:
+                dumped_output = _JSON_OUTPUT_ADAPTER.dump_python(output, mode="json")
+            if not isinstance(dumped_output, Mapping):
+                raise UserError("Function tool output schema requires a JSON object.")
+            converted_output = json.dumps(
+                dict(dumped_output),
+                ensure_ascii=False,
+                separators=(",", ":"),
+            )
+        elif isinstance(output, str):
+            converted_output = output
+        elif _is_programmatic_tool_call(tool_call):
+            structured_output = cls._convert_tool_output_as_structured(output)
+            if structured_output is not None:
+                converted_output = structured_output
+            else:
+                try:
+                    converted_output = _JSON_OUTPUT_ADAPTER.dump_json(output).decode("utf-8")
+                except Exception as error:
+                    raise UserError(
+                        "Programmatic function tool outputs must be strings, structured tool "
+                        "outputs, or JSON-serializable values."
+                    ) from error
+        else:
+            converted_output = cls._convert_tool_output(output)
 
-        return {
+        output_item: FunctionCallOutput = {
             "call_id": tool_call.call_id,
             "output": converted_output,
             "type": "function_call_output",
         }
+        return cast(FunctionCallOutput, cls.copy_tool_call_caller(tool_call, output_item))
+
+    @classmethod
+    def copy_tool_call_caller(
+        cls,
+        tool_call: Any,
+        output_item: Any,
+    ) -> Any:
+        """Copy a program caller relationship from a tool call to its output item."""
+        caller = (
+            tool_call.get("caller")
+            if isinstance(tool_call, Mapping)
+            else getattr(tool_call, "caller", None)
+        )
+        if caller is not None:
+            model_dump = getattr(caller, "model_dump", None)
+            output_item["caller"] = (
+                model_dump(mode="json", exclude_none=True)
+                if callable(model_dump)
+                else _to_dump_compatible(caller)
+            )
+        return output_item
 
     @classmethod
     def _convert_tool_output(cls, output: Any) -> str | ResponseFunctionCallOutputItemListParam:
         """Converts a tool return value into an output acceptable by the Responses API."""
 
+        structured_output = cls._convert_tool_output_as_structured(output)
+        return structured_output if structured_output is not None else str(output)
+
+    @classmethod
+    def _convert_tool_output_as_structured(
+        cls,
+        output: Any,
+    ) -> ResponseFunctionCallOutputItemListParam | None:
+        """Convert known structured tool outputs without stringifying other values."""
+
         # If the output is either a single or list of the known structured output types, convert to
-        # ResponseFunctionCallOutputItemListParam. Else, just stringify.
+        # ResponseFunctionCallOutputItemListParam.
         if isinstance(output, list | tuple):
             maybe_converted_output_list = [
                 cls._maybe_get_output_as_structured_function_output(item) for item in output
@@ -821,14 +920,12 @@ def _convert_tool_output(cls, output: Any) -> str | ResponseFunctionCallOutputIt
                     for item in maybe_converted_output_list
                     if item is not None
                 ]
-            else:
-                return str(output)
-        else:
-            maybe_converted_output = cls._maybe_get_output_as_structured_function_output(output)
-            if maybe_converted_output:
-                return [cls._convert_single_tool_output_pydantic_model(maybe_converted_output)]
-            else:
-                return str(output)
+            return None
+
+        maybe_converted_output = cls._maybe_get_output_as_structured_function_output(output)
+        if maybe_converted_output:
+            return [cls._convert_single_tool_output_pydantic_model(maybe_converted_output)]
+        return None
 
     @classmethod
     def _maybe_get_output_as_structured_function_output(

src/agents/models/chatcmpl_converter.py

@@ -88,7 +88,7 @@ def convert_tool_choice(
         else:
             ensure_tool_choice_supports_backend(
                 tool_choice,
-                backend_name="OpenAI Responses models",
+                backend_name="Chat Completions-compatible models",
             )
             return {
                 "type": "function",