from __future__ import annotations

import asyncio

import json

import sys

from collections.abc import Awaitable, Coroutine

from dataclasses import dataclass

from typing import Annotated, Any, Literal, cast

import pytest

from openai.types.responses import (

ResponseApplyPatchToolCall,

ResponseCustomToolCall,

ResponseFileSearchToolCall,

ResponseFunctionShellToolCall,

ResponseFunctionShellToolCallOutput,

ResponseFunctionToolCall,

ResponseFunctionWebSearch,

ResponseToolSearchCall,

ResponseToolSearchOutputItem,

)

from openai.types.responses.response_apply_patch_tool_call import OperationCreateFile

from openai.types.responses.response_code_interpreter_tool_call import (

ResponseCodeInterpreterToolCall,

)

from openai.types.responses.response_function_shell_tool_call import Action

from openai.types.responses.response_function_tool_call import CallerProgram

from openai.types.responses.response_function_web_search import ActionSearch

from openai.types.responses.response_output_item import (

ImageGenerationCall,

McpApprovalRequest,

McpCall,

McpListTools,

Program,

ProgramOutput,

)

from pydantic import BaseModel, ConfigDict, Field

from typing_extensions import TypedDict

from agents import (

Agent,

ApplyPatchTool,

CodeInterpreterTool,

CustomTool,

HostedMCPTool,

ModelResponse,

ModelSettings,

ProgrammaticToolCallingTool,

RunConfig,

RunItem,

Runner,

RunState,

ShellTool,

ToolCallItem,

ToolCallOutputItem,

ToolExecutionConfig,

ToolGuardrailFunctionOutput,

ToolInputGuardrailData,

ToolOutputGuardrailData,

ToolSearchTool,

Usage,

UserError,

function_tool,

tool_input_guardrail,

tool_output_guardrail,

)

from agents.exceptions import ModelBehaviorError

from agents.items import ItemHelpers

from agents.memory import SQLiteSession

from agents.models.chatcmpl_converter import Converter as ChatCompletionsConverter

from agents.models.openai_responses import Converter as ResponsesConverter

from agents.run_internal.turn_resolution import process_model_response

from agents.tool_context import ToolContext

from .fake_model import FakeModel

from .test_responses import get_handoff_tool_call, get_text_message

PROGRAM_CALL_ID = "call_program"

FUNCTION_CALL_ID = "call_lookup"

PROGRAM_CALLER = {"type": "program", "caller_id": PROGRAM_CALL_ID}

class InventoryOutput(BaseModel):

sku: str

available_units: int

class InventoryAwaitable(Awaitable[InventoryOutput]):

def __await__(self) -> Any:

raise NotImplementedError

class InventoryDict(TypedDict):

sku: str

available_units: int

@dataclass

class InventoryData:

sku: str

available_units: int

class AliasedInventoryOutput(BaseModel):

model_config = ConfigDict(populate_by_name=True)

sku: str

available_units: int = Field(

validation_alias="inputUnits",

serialization_alias="availableUnits",

)

def _program() -> Program:

return Program(

id="program_item",

call_id=PROGRAM_CALL_ID,

code='lookup_inventory(sku="A-1")',

fingerprint="fingerprint",

type="program",

)

def _function_call() -> ResponseFunctionToolCall:

return ResponseFunctionToolCall(

id="function_item",

call_id=FUNCTION_CALL_ID,

name="lookup_inventory",

arguments='{"sku":"A-1"}',

caller=CallerProgram(type="program", caller_id=PROGRAM_CALL_ID),

type="function_call",

)

def _program_output(

status: Literal["completed", "incomplete"] = "completed",

) -> ProgramOutput:

return ProgramOutput(

id="program_output_item",

call_id=PROGRAM_CALL_ID,

result='{"sku":"A-1","available_units":42}',

status=status,

type="program_output",

)

def _hosted_program_call_and_tool(

output_type: str,

allowed_callers: list[Any] | None,

) -> tuple[Any, Any]:

if output_type in ("mcp_approval_request", "mcp_call", "mcp_list_tools"):

mcp_config: dict[str, Any] = {

"type": "mcp",

"server_label": "docs_server",

"server_url": "https://example.com/mcp",

}

if allowed_callers is not None:

mcp_config["allowed_callers"] = allowed_callers

hosted_tool = HostedMCPTool(tool_config=cast(Any, mcp_config))

if output_type == "mcp_list_tools":

return (

McpListTools.model_construct(

id="mcp_list_tools_1",

server_label="docs_server",

tools=[],

type="mcp_list_tools",

caller=PROGRAM_CALLER,

),

hosted_tool,

)

call_type: Any = McpApprovalRequest if output_type == "mcp_approval_request" else McpCall

output = call_type.model_construct(

id=f"{output_type}_1",

arguments="{}",

name="search_docs",

server_label="docs_server",

type=output_type,

caller=PROGRAM_CALLER,

)

return output, hosted_tool

code_interpreter_config: dict[str, Any] = {

"type": "code_interpreter",

"container": "auto",

}

if allowed_callers is not None:

code_interpreter_config["allowed_callers"] = allowed_callers

return (

ResponseCodeInterpreterToolCall.model_construct(

id="code_interpreter_1",

container_id="container_1",

status="completed",

type="code_interpreter_call",

caller=PROGRAM_CALLER,

),

CodeInterpreterTool(tool_config=cast(Any, code_interpreter_config)),

)

def _caller_dict(value: Any) -> dict[str, str]:

if isinstance(value, dict):

return cast(dict[str, str], value)

return cast(dict[str, str], value.model_dump(exclude_none=True))

def _raw_item_type(value: Any) -> str | None:

if isinstance(value, dict):

item_type = value.get("type")

return item_type if isinstance(item_type, str) else None

item_type = getattr(value, "type", None)

return item_type if isinstance(item_type, str) else None

def _function_output_raw_items(result: Any) -> list[dict[str, Any]]:

return [

cast(dict[str, Any], item.raw_item)

for item in result.new_items

if isinstance(item, ToolCallOutputItem)

and isinstance(item.raw_item, dict)

and item.raw_item.get("type") == "function_call_output"

]

def test_responses_converter_serializes_programmatic_tool_configuration() -> None:

@function_tool(allowed_callers=["programmatic"])

def lookup_inventory(sku: str) -> InventoryOutput:

return InventoryOutput(sku=sku, available_units=42)

converted = ResponsesConverter.convert_tools(

tools=[ProgrammaticToolCallingTool(), lookup_inventory],

handoffs=[],

)

assert converted.tools[0] == {"type": "programmatic_tool_calling"}

function_payload = cast(dict[str, Any], converted.tools[1])

assert function_payload["allowed_callers"] == ["programmatic"]

assert function_payload["output_schema"] == lookup_inventory.output_json_schema

assert function_payload["output_schema"] == {

"additionalProperties": False,

"properties": {

"sku": {"title": "Sku", "type": "string"},

"available_units": {"title": "Available Units", "type": "integer"},

},

"required": ["sku", "available_units"],

"title": "InventoryOutput",

"type": "object",

}

assert ResponsesConverter.convert_tool_choice("programmatic_tool_calling") == {

"type": "programmatic_tool_calling"

}

def test_function_tool_infers_typed_dict_and_dataclass_output_schemas() -> None:

@function_tool(allowed_callers=["programmatic"])

def typed_dict_tool() -> InventoryDict:

return {"sku": "A-1", "available_units": 42}

@function_tool(allowed_callers=["programmatic"])

def dataclass_tool() -> InventoryData:

return InventoryData(sku="A-1", available_units=42)

assert typed_dict_tool.output_json_schema is not None

assert typed_dict_tool.output_json_schema["type"] == "object"

assert typed_dict_tool.output_json_schema["additionalProperties"] is False

assert dataclass_tool.output_json_schema is not None

assert dataclass_tool.output_json_schema["type"] == "object"

assert dataclass_tool.output_json_schema["additionalProperties"] is False

@pytest.mark.parametrize(

"return_annotation",

[

Awaitable[InventoryOutput],

Coroutine[Any, Any, InventoryOutput],

InventoryAwaitable,

Awaitable[InventoryOutput] | InventoryOutput,

Awaitable,

Coroutine,

],

)

def test_sync_callable_does_not_infer_through_awaitable_output(

return_annotation: Any,

) -> None:

class LookupInventory:

def __call__(self) -> Any:

raise AssertionError("The handler must not run during tool construction.")

cast(Any, LookupInventory.__call__).__annotations__["return"] = return_annotation

with pytest.raises(UserError, match="programmatic function tool return annotation"):

function_tool(LookupInventory(), allowed_callers=["programmatic"])

@pytest.mark.skipif(sys.version_info < (3, 12), reason="PEP 695 requires Python 3.12")

def test_sync_callable_does_not_infer_through_pep695_awaitable_alias() -> None:

namespace = {

"Any": Any,

"Awaitable": Awaitable,

"InventoryOutput": InventoryOutput,

}

exec(

"type OutputAwaitable = Awaitable[InventoryOutput]\n"

"class LookupInventory:\n"

" def __call__(self) -> OutputAwaitable:\n"

" raise AssertionError\n",

namespace,

)

with pytest.raises(UserError, match="explicit wrapper function"):

function_tool(namespace["LookupInventory"](), allowed_callers=["programmatic"])

def test_function_tool_treats_annotated_plain_returns_as_untyped() -> None:

@function_tool(allowed_callers=["programmatic"])

def string_tool() -> Annotated[str, "plain string"]:

return "ok"

@function_tool(allowed_callers=["programmatic"])

def any_tool() -> Annotated[Any, "untyped value"]:

return {"status": "ok"}

@function_tool(allowed_callers=["programmatic"])

def none_tool() -> Annotated[None, "no value"]:

return None

assert string_tool.output_json_schema is None

assert string_tool._output_type_adapter is None

assert any_tool.output_json_schema is None

assert any_tool._output_type_adapter is None

assert none_tool.output_json_schema is None

assert none_tool._output_type_adapter is None

def test_function_tool_preserves_annotated_structured_return_metadata() -> None:

@function_tool(allowed_callers=["programmatic"])

def lookup_inventory() -> Annotated[

InventoryOutput,

Field(description="Inventory result"),

]:

return InventoryOutput(sku="A-1", available_units=42)

assert lookup_inventory.output_json_schema is not None

assert lookup_inventory.output_json_schema["description"] == "Inventory result"

assert lookup_inventory._output_type_adapter is not None

def test_function_tool_output_type_override_and_raw_schema_are_mutually_exclusive() -> None:

def unannotated_tool() -> Any:

return {"sku": "A-1", "available_units": 42}

tool = function_tool(

unannotated_tool,

allowed_callers=["programmatic"],

output_type=InventoryOutput,

)

assert tool.output_json_schema is not None

assert tool.output_json_schema["title"] == "InventoryOutput"

with pytest.raises(UserError, match="cannot both be provided"):

function_tool(

unannotated_tool,

allowed_callers=["programmatic"],

output_type=InventoryOutput,

output_json_schema={"type": "object"},

)

with pytest.raises(UserError, match="output_type must define a strict JSON object"):

function_tool(

unannotated_tool,

allowed_callers=["programmatic"],

output_type=str,

)

def test_function_tool_rejects_loose_programmatic_output_annotation() -> None:

def loose_dict_tool() -> dict[str, Any]:

return {"sku": "A-1", "available_units": 42}

with pytest.raises(UserError, match="return annotation must define a strict JSON object"):

function_tool(loose_dict_tool, allowed_callers=["programmatic"])

def test_function_tool_does_not_infer_non_programmatic_output() -> None:

@function_tool

def direct_tool() -> InventoryOutput:

return InventoryOutput(sku="A-1", available_units=42)

assert direct_tool.output_json_schema is None

@pytest.mark.parametrize(

"output_json_schema",

[

{"type": "string"},

{"type": "object", "additionalProperties": True},

],

)

def test_function_tool_rejects_non_object_or_non_strict_raw_output_schema(

output_json_schema: dict[str, Any],

) -> None:

with pytest.raises(UserError, match="output_json_schema must define a.*object schema"):

function_tool(

lambda: "ok",

allowed_callers=["programmatic"],

output_json_schema=output_json_schema,

)

@pytest.mark.asyncio

async def test_function_tool_validates_inferred_output_type() -> None:

@function_tool(allowed_callers=["programmatic"])

def invalid_tool() -> InventoryOutput:

return {"sku": "A-1", "available_units": "many"} # type: ignore[return-value]

context = ToolContext(

None,

tool_name=invalid_tool.name,

tool_call_id="invalid",

tool_arguments="{}",

tool_call=_function_call(),

)

with pytest.raises(UserError, match="does not match its declared output type"):

await invalid_tool.on_invoke_tool(context, "{}")

@pytest.mark.asyncio

async def test_schema_backed_programmatic_tool_bypasses_default_failure_formatter() -> None:

@function_tool(allowed_callers=["programmatic"])

def failing_tool() -> InventoryOutput:

raise RuntimeError("inventory unavailable")

context = ToolContext(

None,

tool_name=failing_tool.name,

tool_call_id="failing",

tool_arguments="{}",

tool_call=_function_call(),

)

with pytest.raises(RuntimeError, match="inventory unavailable"):

await failing_tool.on_invoke_tool(context, "{}")

@function_tool(

allowed_callers=["programmatic"],

output_json_schema={

"type": "object",

"properties": {"error": {"type": "string"}},

"required": ["error"],

"additionalProperties": False,

},

)

def failing_declared_schema_tool() -> str:

raise RuntimeError("declared schema unavailable")

declared_context = ToolContext(

None,

tool_name=failing_declared_schema_tool.name,

tool_call_id="failing-declared",

tool_arguments="{}",

tool_call=_function_call(),

)

with pytest.raises(RuntimeError, match="declared schema unavailable"):

await failing_declared_schema_tool.on_invoke_tool(declared_context, "{}")

@pytest.mark.asyncio

async def test_schema_backed_direct_tool_preserves_argument_error_formatter() -> None:

@function_tool(allowed_callers=["direct", "programmatic"])

def failing_tool(sku: str) -> InventoryOutput:

return InventoryOutput(sku=sku, available_units=42)

direct_call = ResponseFunctionToolCall(

id="function_item",

call_id=FUNCTION_CALL_ID,

name=failing_tool.name,

arguments="{}",

type="function_call",

)

context = ToolContext(

None,

tool_name=failing_tool.name,

tool_call_id=FUNCTION_CALL_ID,

tool_arguments="{}",

tool_call=direct_call,

)

result = await failing_tool.on_invoke_tool(context, "{}")

assert result.startswith("An error occurred while running the tool. Please try again. Error:")

assert "sku" in result

@pytest.mark.asyncio

async def test_runner_preserves_direct_error_for_schema_backed_tool() -> None:

model = FakeModel()

direct_call = ResponseFunctionToolCall(

id="function_item",

call_id=FUNCTION_CALL_ID,

name="lookup_inventory",

arguments='{"sku":"A-1"}',

type="function_call",

)

model.add_multiple_turn_outputs([[direct_call], [get_text_message("inventory lookup failed")]])

@function_tool(allowed_callers=["direct", "programmatic"])

def lookup_inventory(sku: str) -> InventoryOutput:

raise RuntimeError(f"inventory unavailable for {sku}")

result = await Runner.run(

Agent(name="inventory", model=model, tools=[lookup_inventory]),

"Check inventory",

)

function_output = next(

item for item in result.new_items if isinstance(item, ToolCallOutputItem)

)

expected_error = (

"An error occurred while running the tool. Please try again. "

"Error: inventory unavailable for A-1"

)

assert result.final_output == "inventory lookup failed"

assert function_output.output == expected_error

assert cast(dict[str, Any], function_output.raw_item)["output"] == expected_error

@pytest.mark.asyncio

async def test_runner_preserves_direct_default_timeout_for_schema_backed_tool() -> None:

model = FakeModel()

direct_call = ResponseFunctionToolCall(

id="function_item",

call_id=FUNCTION_CALL_ID,

name="lookup_inventory",

arguments='{"sku":"A-1"}',

type="function_call",

)

model.add_multiple_turn_outputs([[direct_call], [get_text_message("timed out")]])

@function_tool(allowed_callers=["direct", "programmatic"], timeout=0.01)

async def lookup_inventory(sku: str) -> InventoryOutput:

await asyncio.sleep(0.2)

return InventoryOutput(sku=sku, available_units=42)

result = await Runner.run(

Agent(name="inventory", model=model, tools=[lookup_inventory]),

"Check inventory",

)

function_output = next(

item for item in result.new_items if isinstance(item, ToolCallOutputItem)

)

assert result.final_output == "timed out"

assert isinstance(function_output.output, str)

assert "timed out" in function_output.output.lower()

assert cast(dict[str, Any], function_output.raw_item)["output"] == function_output.output

@pytest.mark.asyncio

async def test_schema_backed_function_tool_accepts_conforming_custom_error_output() -> None:

@function_tool(

allowed_callers=["programmatic"],

failure_error_function=lambda _context, _error: json.dumps(

{"sku": "ERROR", "available_units": 0}

),

)

def failing_tool() -> InventoryOutput:

raise RuntimeError("inventory unavailable")

context = ToolContext(

None,

tool_name=failing_tool.name,

tool_call_id="failing",

tool_arguments="{}",

)

result = await failing_tool.on_invoke_tool(context, "{}")

output = ItemHelpers.tool_call_output_item(

_function_call(),

result,

output_json_schema=failing_tool.output_json_schema,

output_type_adapter=failing_tool._output_type_adapter,

)

assert json.loads(cast(str, output["output"])) == {

"sku": "ERROR",

"available_units": 0,

}

@pytest.mark.asyncio

async def test_schema_backed_programmatic_tool_accepts_conforming_custom_timeout_output() -> None:

model = FakeModel()

model.add_multiple_turn_outputs(

[

[_program(), _function_call()],

[_program_output(), get_text_message("timeout handled")],

]

)

@function_tool(

allowed_callers=["programmatic"],

timeout=0.01,

timeout_error_function=lambda _context, _error: json.dumps(

{"sku": "TIMEOUT", "available_units": 0}

),

)

async def lookup_inventory(sku: str) -> InventoryOutput:

await asyncio.sleep(0.2)

return InventoryOutput(sku=sku, available_units=42)

result = await Runner.run(

Agent(

name="inventory",

model=model,

tools=[ProgrammaticToolCallingTool(), lookup_inventory],

),

"Check inventory",

)

function_output = next(

item for item in result.new_items if isinstance(item, ToolCallOutputItem)

)

assert result.final_output == "timeout handled"

assert json.loads(cast(str, cast(dict[str, Any], function_output.raw_item)["output"])) == {

"sku": "TIMEOUT",

"available_units": 0,

}

def test_schema_backed_function_output_rejects_plain_error_text() -> None:

@function_tool(allowed_callers=["programmatic"])

def lookup_inventory() -> InventoryOutput:

return InventoryOutput(sku="A-1", available_units=42)

with pytest.raises(UserError, match="does not match its declared output schema"):

ItemHelpers.tool_call_output_item(

_function_call(),

"inventory unavailable",

output_json_schema=lookup_inventory.output_json_schema,

output_type_adapter=lookup_inventory._output_type_adapter,

)

with pytest.raises(UserError, match="requires a JSON object"):

ItemHelpers.tool_call_output_item(

_function_call(),

"inventory unavailable",

output_json_schema={"type": "object"},

)

@pytest.mark.asyncio

async def test_function_tool_serializes_typed_output_with_schema_aliases() -> None:

@function_tool(allowed_callers=["programmatic"])

def aliased_tool() -> AliasedInventoryOutput:

return AliasedInventoryOutput(sku="A-1", available_units=42)

context = ToolContext(

None,

tool_name=aliased_tool.name,

tool_call_id="aliased",

tool_arguments="{}",

)

result = await aliased_tool.on_invoke_tool(context, "{}")

output = ItemHelpers.tool_call_output_item(

_function_call(),

result,

output_json_schema=aliased_tool.output_json_schema,

output_type_adapter=aliased_tool._output_type_adapter,

)

assert aliased_tool.output_json_schema is not None

assert "availableUnits" in aliased_tool.output_json_schema["properties"]

assert json.loads(cast(str, output["output"])) == {

"sku": "A-1",

"availableUnits": 42,

}

def test_responses_converter_serializes_allowed_callers_for_other_eligible_tools() -> None:

async def shell_executor(_request: Any) -> str:

return "ok"

def custom_executor(_context: Any, _input: str) -> str:

return "ok"

class Editor:

def create_file(self, _operation: Any) -> str:

return "ok"

def update_file(self, _operation: Any) -> str:

return "ok"

def delete_file(self, _operation: Any) -> str:

return "ok"

converted = ResponsesConverter.convert_tools(

tools=[

ProgrammaticToolCallingTool(),

ShellTool(executor=shell_executor, allowed_callers=["programmatic"]),

ApplyPatchTool(editor=Editor(), allowed_callers=["direct", "programmatic"]),

CustomTool(

name="custom",

description="Custom tool",

on_invoke_tool=custom_executor,

allowed_callers=["programmatic"],

),

],

handoffs=[],

)

tool_payloads = [cast(dict[str, Any], tool) for tool in converted.tools]

assert tool_payloads[1]["allowed_callers"] == ["programmatic"]

assert tool_payloads[2]["allowed_callers"] == ["direct", "programmatic"]

assert tool_payloads[3]["allowed_callers"] == ["programmatic"]

@pytest.mark.parametrize(

"allowed_callers",

[

[],

["direct", "direct"],

["unsupported"],

],

)

def test_tool_construction_rejects_invalid_allowed_callers(

allowed_callers: list[Any],

) -> None:

with pytest.raises(UserError, match="allowed_callers"):

function_tool(lambda: "ok", allowed_callers=allowed_callers)

with pytest.raises(UserError, match="allowed_callers"):

ShellTool(executor=lambda _request: "ok", allowed_callers=allowed_callers)

with pytest.raises(UserError, match="allowed_callers"):

HostedMCPTool(

tool_config=cast(

Any,

{

"type": "mcp",

"server_label": "inventory",

"server_url": "https://example.com/mcp",

"allowed_callers": allowed_callers,

},

)

)

with pytest.raises(UserError, match="allowed_callers"):

CodeInterpreterTool(

tool_config=cast(

Any,

{

"type": "code_interpreter",

"container": "auto",

"allowed_callers": allowed_callers,

},

)

)

def test_responses_converter_rejects_incomplete_programmatic_configuration() -> None:

@function_tool(allowed_callers=["programmatic"])

def programmatic_only() -> str:

return "ok"

with pytest.raises(UserError, match="requires ProgrammaticToolCallingTool"):

ResponsesConverter.convert_tools(tools=[programmatic_only], handoffs=[])

with pytest.raises(UserError, match="requires ProgrammaticToolCallingTool"):

ResponsesConverter.convert_tools(

tools=[],

handoffs=[],

tool_choice="programmatic_tool_calling",

)

with pytest.raises(UserError, match="requires at least one tool"):

ResponsesConverter.convert_tools(

tools=[ProgrammaticToolCallingTool()],

handoffs=[],

)

with pytest.raises(UserError, match="Only one ProgrammaticToolCallingTool"):

ResponsesConverter.convert_tools(

tools=[ProgrammaticToolCallingTool(), ProgrammaticToolCallingTool()],

handoffs=[],

)

def test_responses_converter_accepts_mixed_or_tool_search_managed_configuration() -> None:

@function_tool(allowed_callers=["direct", "programmatic"])

def mixed_callers() -> str:

return "ok"

converted_mixed = ResponsesConverter.convert_tools(tools=[mixed_callers], handoffs=[])

assert cast(dict[str, Any], converted_mixed.tools[0])["allowed_callers"] == [

"direct",

"programmatic",

]

converted_search = ResponsesConverter.convert_tools(

tools=[ProgrammaticToolCallingTool(), ToolSearchTool()],

handoffs=[],

allow_opaque_tool_search_surface=True,

)

assert converted_search.tools == [

{"type": "programmatic_tool_calling"},

{"type": "tool_search"},

]

def test_chat_completions_rejects_programmatic_tool_configuration() -> None:

@function_tool(allowed_callers=["programmatic"])

def lookup_inventory() -> InventoryOutput:

return InventoryOutput(sku="A-1", available_units=42)

with pytest.raises(UserError, match="only supported with OpenAI Responses models"):

ChatCompletionsConverter.tool_to_openai(lookup_inventory)

with pytest.raises(UserError, match="programmatic_tool_calling"):

ChatCompletionsConverter.convert_tool_choice("programmatic_tool_calling")

with pytest.raises(UserError, match="Hosted tools are not supported"):

ChatCompletionsConverter.tool_to_openai(ProgrammaticToolCallingTool())

def test_function_output_preserves_caller_and_uses_declared_json_schema() -> None:

output = ItemHelpers.tool_call_output_item(

_function_call(),

{"sku": "A-1", "available_units": 42},

output_json_schema={"type": "object"},

)

assert json.loads(cast(str, output["output"])) == {

"sku": "A-1",

"available_units": 42,

}

assert _caller_dict(output["caller"]) == PROGRAM_CALLER

programmatic_output = ItemHelpers.tool_call_output_item(

_function_call(),

{"sku": "A-1", "units": [1, 2]},

)

assert json.loads(cast(str, programmatic_output["output"])) == {

"sku": "A-1",

"units": [1, 2],

}

assert _caller_dict(programmatic_output["caller"]) == PROGRAM_CALLER

direct_call = ResponseFunctionToolCall(

id="direct_function_item",

call_id="direct_call",

name="lookup_inventory",

arguments="{}",

type="function_call",

)

legacy_output = ItemHelpers.tool_call_output_item(direct_call, {"sku": "A-1"})

assert legacy_output["output"] == "{'sku': 'A-1'}"

def test_process_model_response_keeps_program_items_in_order() -> None:

@function_tool(allowed_callers=["programmatic"])

def lookup_inventory(sku: str) -> str:

return sku

agent = Agent(

name="inventory",

tools=[ProgrammaticToolCallingTool(), lookup_inventory],

)

response = ModelResponse(

output=[_program(), _function_call(), _program_output("incomplete")],

usage=Usage(),

response_id="response_1",

)

processed = process_model_response(

agent=agent,

all_tools=agent.tools,

response=response,

output_schema=None,

handoffs=[],

)

assert [type(item) for item in processed.new_items] == [

ToolCallItem,

ToolCallItem,

ToolCallOutputItem,

]

assert [_raw_item_type(item.raw_item) for item in processed.new_items] == [

"program",

"function_call",

"program_output",

]

assert processed.tools_used == [

"programmatic_tool_calling",

"lookup_inventory",

"programmatic_tool_calling",

]

@pytest.mark.parametrize("call_id", [None, ""])

def test_process_model_response_rejects_program_without_valid_call_id(

call_id: str | None,

) -> None:

program: dict[str, Any] = {

"type": "program",

"id": "program_item",

"code": "return 42",

"fingerprint": "fingerprint",

}

if call_id is not None:

program["call_id"] = call_id

response = ModelResponse(output=[], usage=Usage(), response_id="response_1")

response.output = cast(Any, [program])

agent = Agent(name="inventory", tools=[ProgrammaticToolCallingTool()])

with pytest.raises(ModelBehaviorError, match="without a valid call_id"):

process_model_response(

agent=agent,

all_tools=agent.tools,

response=response,

output_schema=None,

handoffs=[],

)

@pytest.mark.parametrize(

"program_output",

[_program_output(), _program_output().model_dump(exclude_none=True)],

)

def test_process_model_response_rejects_orphan_program_output(program_output: Any) -> None:

agent = Agent(name="inventory", tools=[ProgrammaticToolCallingTool()])

with pytest.raises(ModelBehaviorError, match="does not match a parent program item"):

process_model_response(

agent=agent,

all_tools=agent.tools,

response=ModelResponse(

output=[program_output],

usage=Usage(),

response_id="response_1",

),

output_schema=None,

handoffs=[],

)

def test_process_model_response_accepts_program_output_for_retained_program() -> None:

agent = Agent(name="inventory", tools=[ProgrammaticToolCallingTool()])

existing_program = ToolCallItem(raw_item=_program(), agent=agent)

processed = process_model_response(

agent=agent,

all_tools=agent.tools,

response=ModelResponse(

output=[_program_output()],

usage=Usage(),

response_id="response_1",

),

output_schema=None,

handoffs=[],

existing_items=[existing_program],

)

assert len(processed.new_items) == 1

assert isinstance(processed.new_items[0], ToolCallOutputItem)

@pytest.mark.parametrize(

("field", "value", "remove_field", "error_match"),

[

("status", None, True, "without a valid status"),

("status", "running", False, "without a valid status"),

("result", None, True, "without a string result"),

("result", 42, False, "without a string result"),

],

)

def test_process_model_response_rejects_malformed_program_output(

field: str,

value: Any,

remove_field: bool,

error_match: str,

) -> None:

agent = Agent(name="inventory", tools=[ProgrammaticToolCallingTool()])

program_output = _program_output().model_dump(exclude_none=True)

if remove_field:

program_output.pop(field)

else:

program_output[field] = value

response = ModelResponse(output=[], usage=Usage(), response_id="response_1")

response.output = cast(Any, [_program(), program_output])

with pytest.raises(ModelBehaviorError, match=error_match):

process_model_response(

agent=agent,

all_tools=agent.tools,

response=response,

output_schema=None,

handoffs=[],

)

@pytest.mark.parametrize("parent_location", ["existing_items", "current_response"])