from typing import Any, cast import pytest from mcp import Tool as MCPTool from openai._models import construct_type from openai.types.responses import ( ResponseApplyPatchToolCall, ResponseCompactionItem, ResponseCustomToolCall, ResponseFunctionShellToolCall, ResponseFunctionShellToolCallOutput, ResponseFunctionToolCall, ResponseOutputItem, ResponseToolSearchCall, ResponseToolSearchOutputItem, ) from openai.types.responses.response_output_item import McpCall, McpListTools, McpListToolsTool from agents import ( Agent, ApplyPatchTool, CompactionItem, CustomTool, Handoff, HostedMCPTool, RunConfig, ShellTool, Tool, function_tool, handoff, tool_namespace, ) from agents.exceptions import ModelBehaviorError, UserError from agents.items import ( HandoffCallItem, MCPListToolsItem, ModelResponse, ToolCallItem, ToolCallOutputItem, ToolSearchCallItem, ToolSearchOutputItem, ) from agents.mcp.util import MCPUtil from agents.run_internal import run_loop from agents.usage import Usage from tests.fake_model import FakeModel from tests.mcp.helpers import FakeMCPServer from tests.test_responses import get_function_tool_call from tests.utils.hitl import ( RecordingEditor, make_apply_patch_dict, make_shell_call, ) def _response(output: list[object]) -> ModelResponse: response = ModelResponse(output=[], usage=Usage(), response_id="resp") response.output = output # type: ignore[assignment] return response def _make_hosted_mcp_list_tools(server_label: str, tool_name: str) -> McpListTools: return McpListTools( id=f"list_{server_label}", server_label=server_label, tools=[ McpListToolsTool( name=tool_name, input_schema={}, description="Search the docs.", annotations={"title": "Search Docs"}, ) ], type="mcp_list_tools", ) def test_process_model_response_shell_call_without_tool_raises() -> None: agent = Agent(name="no-shell", model=FakeModel()) shell_call = make_shell_call("shell-1") with pytest.raises(ModelBehaviorError, match="shell tool"): run_loop.process_model_response( agent=agent, all_tools=[], response=_response([shell_call]), output_schema=None, handoffs=[], ) def test_process_model_response_sets_title_for_local_mcp_function_tool() -> None: agent = Agent(name="local-mcp", model=FakeModel()) mcp_tool = MCPTool(name="search_docs", inputSchema={}, description=None, title="Search Docs") function_tool = MCPUtil.to_function_tool( mcp_tool, FakeMCPServer(), convert_schemas_to_strict=False, ) tool_call = ResponseFunctionToolCall( type="function_call", name="search_docs", call_id="call_search_docs", status="completed", arguments="{}", ) processed = run_loop.process_model_response( agent=agent, all_tools=[function_tool], response=_response([tool_call]), output_schema=None, handoffs=[], ) assert len(processed.new_items) == 1 item = processed.new_items[0] assert isinstance(item, ToolCallItem) assert item.description == "Search Docs" assert item.title == "Search Docs" def test_process_model_response_uses_mcp_list_tools_metadata_for_hosted_mcp_calls() -> None: agent = Agent(name="hosted-mcp", model=FakeModel()) hosted_tool = HostedMCPTool( tool_config=cast( Any, { "type": "mcp", "server_label": "docs_server", "server_url": "https://example.com/mcp", }, ) ) existing_items = [ MCPListToolsItem( agent=agent, raw_item=_make_hosted_mcp_list_tools("docs_server", "search_docs"), ) ] mcp_call = McpCall( id="mcp_call_1", arguments="{}", name="search_docs", server_label="docs_server", type="mcp_call", status="completed", ) processed = run_loop.process_model_response( agent=agent, all_tools=[hosted_tool], response=_response([mcp_call]), output_schema=None, handoffs=[], existing_items=existing_items, ) assert len(processed.new_items) == 1 item = processed.new_items[0] assert isinstance(item, ToolCallItem) assert item.description == "Search the docs." assert item.title == "Search Docs" def test_process_model_response_skips_local_shell_execution_for_hosted_environment() -> None: shell_tool = ShellTool(environment={"type": "container_auto"}) agent = Agent(name="hosted-shell", model=FakeModel(), tools=[shell_tool]) shell_call = make_shell_call("shell-hosted-1") processed = run_loop.process_model_response( agent=agent, all_tools=[shell_tool], response=_response([shell_call]), output_schema=None, handoffs=[], ) assert len(processed.new_items) == 1 assert isinstance(processed.new_items[0], ToolCallItem) assert processed.shell_calls == [] assert processed.tools_used == ["shell"] def test_process_model_response_sanitizes_shell_call_model_object() -> None: shell_call = ResponseFunctionShellToolCall( type="shell_call", id="sh_call_2", call_id="call_shell_2", status="completed", created_by="server", action=cast(Any, {"commands": ["echo hi"], "timeout_ms": 1000}), ) shell_tool = ShellTool(environment={"type": "container_auto"}) agent = Agent(name="hosted-shell-model", model=FakeModel(), tools=[shell_tool]) processed = run_loop.process_model_response( agent=agent, all_tools=[shell_tool], response=_response([shell_call]), output_schema=None, handoffs=[], ) assert len(processed.new_items) == 1 item = processed.new_items[0] assert isinstance(item, ToolCallItem) assert isinstance(item.raw_item, dict) assert item.raw_item["type"] == "shell_call" assert "created_by" not in item.raw_item next_input = item.to_input_item() assert isinstance(next_input, dict) assert next_input["type"] == "shell_call" assert "created_by" not in next_input assert processed.shell_calls == [] assert processed.tools_used == ["shell"] def test_process_model_response_preserves_shell_call_output() -> None: shell_output = { "type": "shell_call_output", "id": "sh_out_1", "call_id": "call_shell_1", "status": "completed", "max_output_length": 1000, "output": [ { "stdout": "ok\n", "stderr": "", "outcome": {"type": "exit", "exit_code": 0}, } ], } agent = Agent(name="shell-output", model=FakeModel()) processed = run_loop.process_model_response( agent=agent, all_tools=[], response=_response([shell_output]), output_schema=None, handoffs=[], ) assert len(processed.new_items) == 1 assert isinstance(processed.new_items[0], ToolCallOutputItem) assert processed.new_items[0].raw_item == shell_output assert processed.tools_used == ["shell"] assert processed.shell_calls == [] def test_process_model_response_sanitizes_shell_call_output_model_object() -> None: shell_output = ResponseFunctionShellToolCallOutput( type="shell_call_output", id="sh_out_2", call_id="call_shell_2", status="completed", created_by="server", output=cast( Any, [ { "stdout": "ok\n", "stderr": "", "outcome": {"type": "exit", "exit_code": 0}, "created_by": "server", } ], ), ) agent = Agent(name="shell-output-model", model=FakeModel()) processed = run_loop.process_model_response( agent=agent, all_tools=[], response=_response([shell_output]), output_schema=None, handoffs=[], ) assert len(processed.new_items) == 1 item = processed.new_items[0] assert isinstance(item, ToolCallOutputItem) assert isinstance(item.raw_item, dict) assert item.raw_item["type"] == "shell_call_output" assert "created_by" not in item.raw_item shell_outputs = item.raw_item.get("output") assert isinstance(shell_outputs, list) assert isinstance(shell_outputs[0], dict) assert "created_by" not in shell_outputs[0] next_input = item.to_input_item() assert isinstance(next_input, dict) assert next_input["type"] == "shell_call_output" assert "status" not in next_input assert "created_by" not in next_input next_outputs = next_input.get("output") assert isinstance(next_outputs, list) assert isinstance(next_outputs[0], dict) assert "created_by" not in next_outputs[0] assert processed.tools_used == ["shell"] def test_process_model_response_apply_patch_call_without_tool_raises() -> None: agent = Agent(name="no-apply", model=FakeModel()) apply_patch_call = make_apply_patch_dict("apply-1", diff="-old\n+new\n") with pytest.raises(ModelBehaviorError, match="apply_patch tool"): run_loop.process_model_response( agent=agent, all_tools=[], response=_response([apply_patch_call]), output_schema=None, handoffs=[], ) def test_process_model_response_sanitizes_apply_patch_call_model_object() -> None: editor = RecordingEditor() apply_patch_tool = ApplyPatchTool(editor=editor) agent = Agent(name="apply-agent-model", model=FakeModel(), tools=[apply_patch_tool]) apply_patch_call = ResponseApplyPatchToolCall( type="apply_patch_call", id="ap_call_1", call_id="call_apply_1", status="completed", created_by="server", operation=cast( Any, {"type": "update_file", "path": "test.md", "diff": "-old\n+new\n"}, ), ) processed = run_loop.process_model_response( agent=agent, all_tools=[apply_patch_tool], response=_response([apply_patch_call]), output_schema=None, handoffs=[], ) assert len(processed.new_items) == 1 item = processed.new_items[0] assert isinstance(item, ToolCallItem) assert isinstance(item.raw_item, dict) assert item.raw_item["type"] == "apply_patch_call" assert "created_by" not in item.raw_item next_input = item.to_input_item() assert isinstance(next_input, dict) assert next_input["type"] == "apply_patch_call" assert "created_by" not in next_input assert len(processed.apply_patch_calls) == 1 queued_call = processed.apply_patch_calls[0].tool_call assert isinstance(queued_call, dict) assert queued_call["type"] == "apply_patch_call" assert "created_by" not in queued_call assert processed.tools_used == [apply_patch_tool.name] def test_process_model_response_queues_apply_patch_call() -> None: editor = RecordingEditor() apply_patch_tool = ApplyPatchTool(editor=editor) agent = Agent(name="apply-agent", model=FakeModel(), tools=[apply_patch_tool]) apply_patch_call = make_apply_patch_dict("apply-1") processed = run_loop.process_model_response( agent=agent, all_tools=[apply_patch_tool], response=_response([apply_patch_call]), output_schema=None, handoffs=[], ) assert processed.apply_patch_calls, "apply_patch call should be queued" converted_call = processed.apply_patch_calls[0].tool_call assert isinstance(converted_call, dict) assert converted_call.get("type") == "apply_patch_call" def test_process_model_response_queues_hosted_apply_patch_from_custom_tool_call() -> None: editor = RecordingEditor() apply_patch_tool = ApplyPatchTool(editor=editor) agent = Agent(name="apply-agent-custom", model=FakeModel(), tools=[apply_patch_tool]) custom_call = ResponseCustomToolCall( type="custom_tool_call", name="apply_patch", call_id="custom-apply-1", input='{"type":"update_file","path":"test.md","diff":"-old\\n+new\\n"}', ) processed = run_loop.process_model_response( agent=agent, all_tools=[apply_patch_tool], response=_response([custom_call]), output_schema=None, handoffs=[], ) assert len(processed.new_items) == 1 item = processed.new_items[0] assert isinstance(item, ToolCallItem) assert isinstance(item.raw_item, dict) assert item.raw_item["type"] == "apply_patch_call" assert processed.apply_patch_calls, "apply_patch call should be queued" converted_call = processed.apply_patch_calls[0].tool_call assert isinstance(converted_call, dict) assert converted_call["type"] == "apply_patch_call" assert converted_call["operation"]["type"] == "update_file" assert processed.tools_used == [apply_patch_tool.name] def test_process_model_response_queues_custom_tool_call_for_custom_tool() -> None: custom_tool = CustomTool( name="raw_editor", description="Edit raw text.", on_invoke_tool=lambda _ctx, raw_input: raw_input, format={"type": "text"}, ) agent = Agent(name="custom-agent", model=FakeModel(), tools=[custom_tool]) custom_call = ResponseCustomToolCall( type="custom_tool_call", name="raw_editor", call_id="custom-apply-1", input="-old\n+new\n", ) processed = run_loop.process_model_response( agent=agent, all_tools=[custom_tool], response=_response([custom_call]), output_schema=None, handoffs=[], ) item = processed.new_items[0] assert isinstance(item, ToolCallItem) assert cast(object, item.raw_item) is custom_call assert processed.apply_patch_calls == [] assert processed.custom_tool_calls[0].tool_call is custom_call assert processed.custom_tool_calls[0].custom_tool is custom_tool def test_process_model_response_prefers_namespaced_function_over_apply_patch_fallback() -> None: namespaced_tool = tool_namespace( name="billing", description="Billing tools", tools=[function_tool(lambda payload: payload, name_override="apply_patch_lookup")], )[0] all_tools: list[Tool] = [namespaced_tool] agent = Agent(name="billing-agent", model=FakeModel(), tools=all_tools) processed = run_loop.process_model_response( agent=agent, all_tools=all_tools, response=_response( [ get_function_tool_call( "apply_patch_lookup", '{"payload":"value"}', namespace="billing", ) ] ), output_schema=None, handoffs=[], ) assert len(processed.functions) == 1 assert processed.functions[0].function_tool is namespaced_tool assert processed.apply_patch_calls == [] def test_process_model_response_handles_compaction_item() -> None: agent = Agent(name="compaction-agent", model=FakeModel()) compaction_item = ResponseCompactionItem( id="comp-1", encrypted_content="enc", type="compaction", created_by="server", ) processed = run_loop.process_model_response( agent=agent, all_tools=[], response=_response([compaction_item]), output_schema=None, handoffs=[], ) assert len(processed.new_items) == 1 item = processed.new_items[0] assert isinstance(item, CompactionItem) assert isinstance(item.raw_item, dict) assert item.raw_item["type"] == "compaction" assert item.raw_item["encrypted_content"] == "enc" assert "created_by" not in item.raw_item def test_process_model_response_classifies_tool_search_items() -> None: agent = Agent(name="tool-search-agent", model=FakeModel()) tool_search_call = construct_type( type_=ResponseOutputItem, value={ "id": "tsc_123", "type": "tool_search_call", "arguments": {"paths": ["crm"], "query": "profile"}, "execution": "server", "status": "completed", }, ) tool_search_output = construct_type( type_=ResponseOutputItem, value={ "id": "tso_123", "type": "tool_search_output", "execution": "server", "status": "completed", "tools": [ { "type": "function", "name": "get_customer_profile", "description": "Fetch a CRM customer profile.", "parameters": { "type": "object", "properties": { "customer_id": { "type": "string", } }, "required": ["customer_id"], }, "defer_loading": True, } ], }, ) processed = run_loop.process_model_response( agent=agent, all_tools=[], response=_response([tool_search_call, tool_search_output]), output_schema=None, handoffs=[], ) assert isinstance(processed.new_items[0], ToolSearchCallItem) assert isinstance(processed.new_items[0].raw_item, ResponseToolSearchCall) assert isinstance(processed.new_items[1], ToolSearchOutputItem) assert isinstance(processed.new_items[1].raw_item, ResponseToolSearchOutputItem) assert processed.tools_used == ["tool_search", "tool_search"] def test_process_model_response_uses_namespace_for_duplicate_function_names() -> None: crm_tool = function_tool(lambda customer_id: customer_id, name_override="lookup_account") billing_tool = function_tool(lambda customer_id: customer_id, name_override="lookup_account") crm_namespace = tool_namespace( name="crm", description="CRM tools", tools=[crm_tool], ) billing_namespace = tool_namespace( name="billing", description="Billing tools", tools=[billing_tool], ) all_tools: list[Tool] = [*crm_namespace, *billing_namespace] agent = Agent(name="billing-agent", model=FakeModel(), tools=all_tools) processed = run_loop.process_model_response( agent=agent, all_tools=all_tools, response=_response( [ get_function_tool_call( "lookup_account", '{"customer_id":"customer_42"}', namespace="billing", ) ] ), output_schema=None, handoffs=[], ) assert len(processed.functions) == 1 assert processed.functions[0].function_tool is billing_namespace[0] assert processed.tools_used == ["billing.lookup_account"] def test_process_model_response_collapses_synthetic_deferred_namespace_in_tools_used() -> None: deferred_tool = function_tool( lambda city: city, name_override="get_weather", defer_loading=True, ) agent = Agent(name="weather-agent", model=FakeModel(), tools=[deferred_tool]) processed = run_loop.process_model_response( agent=agent, all_tools=[deferred_tool], response=_response( [ get_function_tool_call( "get_weather", '{"city":"Tokyo"}', namespace="get_weather", ) ] ), output_schema=None, handoffs=[], ) assert len(processed.functions) == 1 assert processed.functions[0].function_tool is deferred_tool assert processed.tools_used == ["get_weather"] def test_process_model_response_rejects_bare_name_for_duplicate_namespaced_functions() -> None: crm_tool = function_tool(lambda customer_id: customer_id, name_override="lookup_account") billing_tool = function_tool(lambda customer_id: customer_id, name_override="lookup_account") crm_namespace = tool_namespace( name="crm", description="CRM tools", tools=[crm_tool], ) billing_namespace = tool_namespace( name="billing", description="Billing tools", tools=[billing_tool], ) all_tools: list[Tool] = [*crm_namespace, *billing_namespace] agent = Agent(name="billing-agent", model=FakeModel(), tools=all_tools) with pytest.raises(ModelBehaviorError, match="Tool lookup_account not found"): run_loop.process_model_response( agent=agent, all_tools=all_tools, response=_response( [get_function_tool_call("lookup_account", '{"customer_id":"customer_42"}')] ), output_schema=None, handoffs=[], ) def test_process_model_response_uses_last_duplicate_top_level_function() -> None: first_tool = function_tool(lambda customer_id: f"first:{customer_id}", name_override="lookup") second_tool = function_tool(lambda customer_id: f"second:{customer_id}", name_override="lookup") all_tools: list[Tool] = [first_tool, second_tool] agent = Agent(name="lookup-agent", model=FakeModel(), tools=all_tools) processed = run_loop.process_model_response( agent=agent, all_tools=all_tools, response=_response([get_function_tool_call("lookup", '{"customer_id":"customer_42"}')]), output_schema=None, handoffs=[], ) assert len(processed.functions) == 1 assert processed.functions[0].function_tool is second_tool def test_process_model_response_rejects_reserved_same_name_namespace_shape() -> None: invalid_tool = function_tool(lambda customer_id: customer_id, name_override="lookup_account") invalid_tool._tool_namespace = "lookup_account" invalid_tool._tool_namespace_description = "Same-name namespace" all_tools: list[Tool] = [invalid_tool] agent = Agent(name="lookup-agent", model=FakeModel(), tools=all_tools) with pytest.raises(UserError, match="synthetic namespace `lookup_account.lookup_account`"): run_loop.process_model_response( agent=agent, all_tools=all_tools, response=_response( [ get_function_tool_call( "lookup_account", '{"customer_id":"customer_42"}', namespace="lookup_account", ) ] ), output_schema=None, handoffs=[], ) def test_process_model_response_rejects_qualified_name_collision_with_dotted_top_level_tool() -> ( None ): dotted_top_level_tool = function_tool( lambda customer_id: customer_id, name_override="crm.lookup_account", ) namespaced_tool = tool_namespace( name="crm", description="CRM tools", tools=[function_tool(lambda customer_id: customer_id, name_override="lookup_account")], )[0] all_tools: list[Tool] = [dotted_top_level_tool, namespaced_tool] agent = Agent(name="lookup-agent", model=FakeModel(), tools=all_tools) with pytest.raises(UserError, match="qualified name `crm.lookup_account`"): run_loop.process_model_response( agent=agent, all_tools=all_tools, response=_response( [ get_function_tool_call( "lookup_account", '{"customer_id":"customer_42"}', namespace="crm", ) ] ), output_schema=None, handoffs=[], ) def test_process_model_response_prefers_visible_top_level_function_over_deferred_same_name_tool(): visible_tool = function_tool( lambda customer_id: f"visible:{customer_id}", name_override="lookup_account", ) deferred_tool = function_tool( lambda customer_id: f"deferred:{customer_id}", name_override="lookup_account", defer_loading=True, ) all_tools: list[Tool] = [visible_tool, deferred_tool] agent = Agent(name="lookup-agent", model=FakeModel(), tools=all_tools) processed = run_loop.process_model_response( agent=agent, all_tools=all_tools, response=_response( [get_function_tool_call("lookup_account", '{"customer_id":"customer_42"}')] ), output_schema=None, handoffs=[], ) assert len(processed.functions) == 1 assert processed.functions[0].function_tool is visible_tool assert getattr(processed.functions[0].tool_call, "namespace", None) is None assert isinstance(processed.new_items[0], ToolCallItem) assert getattr(processed.new_items[0].raw_item, "namespace", None) is None def test_process_model_response_uses_internal_lookup_key_for_deferred_top_level_calls() -> None: visible_tool = function_tool( lambda customer_id: f"visible:{customer_id}", name_override="lookup_account.lookup_account", ) deferred_tool = function_tool( lambda customer_id: f"deferred:{customer_id}", name_override="lookup_account", defer_loading=True, ) all_tools: list[Tool] = [visible_tool, deferred_tool] agent = Agent(name="lookup-agent", model=FakeModel(), tools=all_tools) processed = run_loop.process_model_response( agent=agent, all_tools=all_tools, response=_response( [ get_function_tool_call( "lookup_account", '{"customer_id":"customer_42"}', namespace="lookup_account", ) ] ), output_schema=None, handoffs=[], ) assert len(processed.functions) == 1 assert processed.functions[0].function_tool is deferred_tool def test_process_model_response_preserves_synthetic_namespace_for_deferred_top_level_tools() -> ( None ): deferred_tool = function_tool( lambda city: city, name_override="get_weather", defer_loading=True, ) all_tools: list[Tool] = [deferred_tool] agent = Agent(name="weather-agent", model=FakeModel(), tools=all_tools) processed = run_loop.process_model_response( agent=agent, all_tools=all_tools, response=_response( [get_function_tool_call("get_weather", '{"city":"Tokyo"}', namespace="get_weather")] ), output_schema=None, handoffs=[], ) assert len(processed.functions) == 1 assert processed.functions[0].function_tool is deferred_tool assert getattr(processed.functions[0].tool_call, "namespace", None) == "get_weather" assert isinstance(processed.new_items[0], ToolCallItem) assert getattr(processed.new_items[0].raw_item, "namespace", None) == "get_weather" def test_process_model_response_prefers_namespaced_function_over_handoff_name_collision() -> None: billing_tool = function_tool(lambda customer_id: customer_id, name_override="lookup_account") billing_namespace = tool_namespace( name="billing", description="Billing tools", tools=[billing_tool], ) handoff_target = Agent(name="lookup-agent", model=FakeModel()) lookup_handoff: Handoff = handoff(handoff_target, tool_name_override="lookup_account") all_tools: list[Tool] = [*billing_namespace] agent = Agent(name="billing-agent", model=FakeModel(), tools=all_tools) processed = run_loop.process_model_response( agent=agent, all_tools=all_tools, response=_response( [ get_function_tool_call( "lookup_account", '{"customer_id":"customer_42"}', namespace="billing", ) ] ), output_schema=None, handoffs=[lookup_handoff], ) assert len(processed.functions) == 1 assert processed.functions[0].function_tool is billing_namespace[0] assert processed.handoffs == [] assert len(processed.new_items) == 1 assert isinstance(processed.new_items[0], ToolCallItem) assert not isinstance(processed.new_items[0], HandoffCallItem) def test_process_model_response_rejects_mismatched_function_namespace() -> None: bare_tool = function_tool(lambda customer_id: customer_id, name_override="lookup_account") all_tools: list[Tool] = [bare_tool] agent = Agent(name="bare-agent", model=FakeModel(), tools=all_tools) with pytest.raises(ModelBehaviorError, match="crm.lookup_account"): run_loop.process_model_response( agent=agent, all_tools=all_tools, response=_response( [ get_function_tool_call( "lookup_account", '{"customer_id":"customer_42"}', namespace="crm", ) ] ), output_schema=None, handoffs=[], ) def test_process_model_response_collects_missing_function_tool_when_opted_in() -> None: agent = Agent(name="test", model=FakeModel(), tools=[function_tool(lambda: "ok")]) missing_call = get_function_tool_call("missing_tool", "{}", call_id="call_missing") processed = run_loop.process_model_response( agent=agent, all_tools=agent.tools, response=_response([missing_call]), output_schema=None, handoffs=[], run_config=RunConfig(tool_not_found_behavior="return_error_to_model"), ) assert len(processed.new_items) == 1 assert isinstance(processed.new_items[0], ToolCallItem) assert processed.functions == [] assert len(processed.function_tools_not_found) == 1 assert processed.function_tools_not_found[0].tool_call is missing_call assert processed.function_tools_not_found[0].tool_name == "missing_tool" assert processed.has_tools_or_approvals_to_run()