import asyncio

import contextlib

import copy

import dataclasses

import json

import logging

import time

from collections.abc import Callable

from typing import Any, cast

import pytest

from pydantic import BaseModel

from typing_extensions import TypedDict

import agents._debug as _debug

import agents.tool as tool_module

from agents import (

Agent,

AgentBase,

FunctionTool,

HostedMCPTool,

ModelBehaviorError,

RunContextWrapper,

ToolGuardrailFunctionOutput,

ToolInputGuardrailData,

ToolOutputGuardrailData,

ToolSearchTool,

ToolTimeoutError,

UserError,

function_tool,

tool_input_guardrail,

tool_namespace,

tool_output_guardrail,

)

from agents.tool import default_tool_error_function

from agents.tool_context import ToolContext

def argless_function() -> str:

return "ok"

def test_tool_namespace_copies_tools_with_metadata() -> None:

tool = function_tool(argless_function)

namespaced_tools = tool_namespace(

name="crm",

description="CRM tools",

tools=[tool],

)

assert len(namespaced_tools) == 1

assert namespaced_tools[0] is not tool

assert namespaced_tools[0]._tool_namespace == "crm"

assert namespaced_tools[0]._tool_namespace_description == "CRM tools"

assert namespaced_tools[0].qualified_name == "crm.argless_function"

assert tool._tool_namespace is None

assert tool.qualified_name == "argless_function"

def test_tool_namespace_requires_keyword_arguments() -> None:

tool = function_tool(argless_function)

with pytest.raises(TypeError):

tool_namespace("crm", "CRM tools", [tool]) # type: ignore[misc]

def test_tool_namespace_requires_non_empty_description() -> None:

tool = function_tool(argless_function)

with pytest.raises(UserError, match="non-empty description"):

tool_namespace(

name="crm",

description=None,

tools=[tool],

)

with pytest.raises(UserError, match="non-empty description"):

tool_namespace(

name="crm",

description=" ",

tools=[tool],

)

def test_tool_namespace_rejects_reserved_same_name_shape() -> None:

tool = function_tool(argless_function, name_override="lookup_account")

with pytest.raises(UserError, match="synthetic namespace `lookup_account.lookup_account`"):

tool_namespace(

name="lookup_account",

description="Same-name namespace",

tools=[tool],

)

@pytest.mark.asyncio

async def test_argless_function():

tool = function_tool(argless_function)

assert tool.name == "argless_function"

result = await tool.on_invoke_tool(

ToolContext(context=None, tool_name=tool.name, tool_call_id="1", tool_arguments=""), ""

)

assert result == "ok"

def argless_with_context(ctx: ToolContext[str]) -> str:

return "ok"

@pytest.mark.asyncio

async def test_argless_with_context():

tool = function_tool(argless_with_context)

assert tool.name == "argless_with_context"

result = await tool.on_invoke_tool(

ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments=""), ""

)

assert result == "ok"

# Extra JSON should not raise an error

result = await tool.on_invoke_tool(

ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments='{"a": 1}'),

'{"a": 1}',

)

assert result == "ok"

def simple_function(a: int, b: int = 5):

return a + b

@pytest.mark.asyncio

async def test_simple_function():

tool = function_tool(simple_function, failure_error_function=None)

assert tool.name == "simple_function"

result = await tool.on_invoke_tool(

ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments='{"a": 1}'),

'{"a": 1}',

)

assert result == 6

result = await tool.on_invoke_tool(

ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments='{"a": 1, "b": 2}'),

'{"a": 1, "b": 2}',

)

assert result == 3

# Missing required argument should raise an error

with pytest.raises(ModelBehaviorError):

await tool.on_invoke_tool(

ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments=""), ""

)

@pytest.mark.asyncio

async def test_sync_function_runs_via_to_thread(monkeypatch: pytest.MonkeyPatch) -> None:

calls = {"to_thread": 0, "func": 0}

def sync_func() -> str:

calls["func"] += 1

return "ok"

async def fake_to_thread(

func: Callable[..., Any],

/,

*args: Any,

**kwargs: Any,

) -> Any:

calls["to_thread"] += 1

return func(*args, **kwargs)

monkeypatch.setattr(asyncio, "to_thread", fake_to_thread)

tool = function_tool(sync_func)

result = await tool.on_invoke_tool(

ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments=""), ""

)

assert result == "ok"

assert calls["to_thread"] == 1

assert calls["func"] == 1

@pytest.mark.asyncio

async def test_sync_function_does_not_block_event_loop() -> None:

def sync_func() -> str:

time.sleep(0.2)

return "ok"

tool = function_tool(sync_func)

async def run_tool() -> Any:

return await tool.on_invoke_tool(

ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments=""), ""

)

tool_task: asyncio.Task[Any] = asyncio.create_task(run_tool())

background_task: asyncio.Task[None] = asyncio.create_task(asyncio.sleep(0.01))

done, pending = await asyncio.wait(

{tool_task, background_task},

return_when=asyncio.FIRST_COMPLETED,

)

try:

assert background_task in done

assert tool_task in pending

assert await tool_task == "ok"

finally:

if not background_task.done():

background_task.cancel()

with contextlib.suppress(asyncio.CancelledError):

await background_task

if not tool_task.done():

tool_task.cancel()

with contextlib.suppress(asyncio.CancelledError):

await tool_task

class Foo(BaseModel):

a: int

b: int = 5

class Bar(TypedDict):

x: str

y: int

def complex_args_function(foo: Foo, bar: Bar, baz: str = "hello"):

return f"{foo.a + foo.b} {bar['x']}{bar['y']} {baz}"

@tool_input_guardrail

def reject_args_guardrail(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:

"""Reject tool calls for test purposes."""

return ToolGuardrailFunctionOutput.reject_content(

message="blocked",

output_info={"tool": data.context.tool_name},

)

@tool_output_guardrail

def allow_output_guardrail(data: ToolOutputGuardrailData) -> ToolGuardrailFunctionOutput:

"""Allow tool outputs for test purposes."""

return ToolGuardrailFunctionOutput.allow(output_info={"echo": data.output})

@pytest.mark.asyncio

async def test_complex_args_function():

tool = function_tool(complex_args_function, failure_error_function=None)

assert tool.name == "complex_args_function"

valid_json = json.dumps(

{

"foo": Foo(a=1).model_dump(),

"bar": Bar(x="hello", y=10),

}

)

result = await tool.on_invoke_tool(

ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments=valid_json),

valid_json,

)

assert result == "6 hello10 hello"

valid_json = json.dumps(

{

"foo": Foo(a=1, b=2).model_dump(),

"bar": Bar(x="hello", y=10),

}

)

result = await tool.on_invoke_tool(

ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments=valid_json),

valid_json,

)

assert result == "3 hello10 hello"

valid_json = json.dumps(

{

"foo": Foo(a=1, b=2).model_dump(),

"bar": Bar(x="hello", y=10),

"baz": "world",

}

)

result = await tool.on_invoke_tool(

ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments=valid_json),

valid_json,

)

assert result == "3 hello10 world"

# Missing required argument should raise an error

with pytest.raises(ModelBehaviorError):

await tool.on_invoke_tool(

ToolContext(

None, tool_name=tool.name, tool_call_id="1", tool_arguments='{"foo": {"a": 1}}'

),

'{"foo": {"a": 1}}',

)

def test_function_config_overrides():

tool = function_tool(simple_function, name_override="custom_name")

assert tool.name == "custom_name"

tool = function_tool(simple_function, description_override="custom description")

assert tool.description == "custom description"

tool = function_tool(

simple_function,

name_override="custom_name",

description_override="custom description",

)

assert tool.name == "custom_name"

assert tool.description == "custom description"

def test_func_schema_is_strict():

tool = function_tool(simple_function)

assert tool.strict_json_schema, "Should be strict by default"

assert (

"additionalProperties" in tool.params_json_schema

and not tool.params_json_schema["additionalProperties"]

)

tool = function_tool(complex_args_function)

assert tool.strict_json_schema, "Should be strict by default"

assert (

"additionalProperties" in tool.params_json_schema

and not tool.params_json_schema["additionalProperties"]

)

@pytest.mark.asyncio

async def test_manual_function_tool_creation_works():

def do_some_work(data: str) -> str:

return f"{data}_done"

class FunctionArgs(BaseModel):

data: str

async def run_function(ctx: RunContextWrapper[Any], args: str) -> str:

parsed = FunctionArgs.model_validate_json(args)

return do_some_work(data=parsed.data)

tool = FunctionTool(

name="test",

description="Processes extracted user data",

params_json_schema=FunctionArgs.model_json_schema(),

on_invoke_tool=run_function,

)

assert tool.name == "test"

assert tool.description == "Processes extracted user data"

for key, value in FunctionArgs.model_json_schema().items():

assert tool.params_json_schema[key] == value

assert tool.strict_json_schema

result = await tool.on_invoke_tool(

ToolContext(

None, tool_name=tool.name, tool_call_id="1", tool_arguments='{"data": "hello"}'

),

'{"data": "hello"}',

)

assert result == "hello_done"

tool_not_strict = FunctionTool(

name="test",

description="Processes extracted user data",

params_json_schema=FunctionArgs.model_json_schema(),

on_invoke_tool=run_function,

strict_json_schema=False,

)

assert not tool_not_strict.strict_json_schema

assert "additionalProperties" not in tool_not_strict.params_json_schema

result = await tool_not_strict.on_invoke_tool(

ToolContext(

None,

tool_name=tool_not_strict.name,

tool_call_id="1",

tool_arguments='{"data": "hello", "bar": "baz"}',

),

'{"data": "hello", "bar": "baz"}',

)

assert result == "hello_done"

@pytest.mark.asyncio

async def test_function_tool_default_error_works():

def my_func(a: int, b: int = 5):

raise ValueError("test")

tool = function_tool(my_func)

ctx = ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments="")

result = await tool.on_invoke_tool(ctx, "")

assert "Invalid JSON" in str(result)

result = await tool.on_invoke_tool(ctx, "{}")

assert "Invalid JSON" in str(result)

result = await tool.on_invoke_tool(ctx, '{"a": 1}')

assert result == default_tool_error_function(ctx, ValueError("test"))

result = await tool.on_invoke_tool(ctx, '{"a": 1, "b": 2}')

assert result == default_tool_error_function(ctx, ValueError("test"))

@pytest.mark.asyncio

async def test_sync_custom_error_function_works():

def my_func(a: int, b: int = 5):

raise ValueError("test")

def custom_sync_error_function(ctx: RunContextWrapper[Any], error: Exception) -> str:

return f"error_{error.__class__.__name__}"

tool = function_tool(my_func, failure_error_function=custom_sync_error_function)

ctx = ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments="")

result = await tool.on_invoke_tool(ctx, "")

assert result == "error_ModelBehaviorError"

result = await tool.on_invoke_tool(ctx, "{}")

assert result == "error_ModelBehaviorError"

result = await tool.on_invoke_tool(ctx, '{"a": 1}')

assert result == "error_ValueError"

result = await tool.on_invoke_tool(ctx, '{"a": 1, "b": 2}')

assert result == "error_ValueError"

@pytest.mark.asyncio

async def test_async_custom_error_function_works():

async def my_func(a: int, b: int = 5):

raise ValueError("test")

def custom_sync_error_function(ctx: RunContextWrapper[Any], error: Exception) -> str:

return f"error_{error.__class__.__name__}"

tool = function_tool(my_func, failure_error_function=custom_sync_error_function)

ctx = ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments="")

result = await tool.on_invoke_tool(ctx, "")

assert result == "error_ModelBehaviorError"

result = await tool.on_invoke_tool(ctx, "{}")

assert result == "error_ModelBehaviorError"

result = await tool.on_invoke_tool(ctx, '{"a": 1}')

assert result == "error_ValueError"

result = await tool.on_invoke_tool(ctx, '{"a": 1, "b": 2}')

assert result == "error_ValueError"

class BoolCtx(BaseModel):

enable_tools: bool

@pytest.mark.asyncio

async def test_is_enabled_bool_and_callable():

@function_tool(is_enabled=False)

def disabled_tool():

return "nope"

async def cond_enabled(ctx: RunContextWrapper[BoolCtx], agent: AgentBase) -> bool:

return ctx.context.enable_tools

@function_tool(is_enabled=cond_enabled)

def another_tool():

return "hi"

async def third_tool_on_invoke_tool(ctx: RunContextWrapper[Any], args: str) -> str:

return "third"

third_tool = FunctionTool(

name="third_tool",

description="third tool",

on_invoke_tool=third_tool_on_invoke_tool,

is_enabled=lambda ctx, agent: ctx.context.enable_tools,

params_json_schema={},

)

agent = Agent(name="t", tools=[disabled_tool, another_tool, third_tool])

context_1 = RunContextWrapper(BoolCtx(enable_tools=False))

context_2 = RunContextWrapper(BoolCtx(enable_tools=True))

tools_with_ctx = await agent.get_all_tools(context_1)

assert tools_with_ctx == []

tools_with_ctx = await agent.get_all_tools(context_2)

assert len(tools_with_ctx) == 2

assert tools_with_ctx[0].name == "another_tool"

assert tools_with_ctx[1].name == "third_tool"

@pytest.mark.asyncio

async def test_get_all_tools_preserves_explicit_tool_search_when_deferred_tools_are_disabled():

async def deferred_enabled(ctx: RunContextWrapper[BoolCtx], agent: AgentBase) -> bool:

return ctx.context.enable_tools

@function_tool(defer_loading=True, is_enabled=deferred_enabled)

def deferred_lookup() -> str:

return "loaded"

agent = Agent(name="t", tools=[deferred_lookup, ToolSearchTool()])

tools_with_disabled_context = await agent.get_all_tools(

RunContextWrapper(BoolCtx(enable_tools=False))

)

assert len(tools_with_disabled_context) == 1

assert isinstance(tools_with_disabled_context[0], ToolSearchTool)

tools_with_enabled_context = await agent.get_all_tools(

RunContextWrapper(BoolCtx(enable_tools=True))

)

assert tools_with_enabled_context[0] is deferred_lookup

assert isinstance(tools_with_enabled_context[1], ToolSearchTool)

@pytest.mark.asyncio

async def test_get_all_tools_keeps_tool_search_for_namespace_only_tools():

namespaced_lookup = tool_namespace(

name="crm",

description="CRM tools",

tools=[function_tool(lambda account_id: account_id, name_override="lookup_account")],

)[0]

agent = Agent(name="t", tools=[namespaced_lookup, ToolSearchTool()])

tools = await agent.get_all_tools(RunContextWrapper(BoolCtx(enable_tools=False)))

assert tools[0] is namespaced_lookup

assert isinstance(tools[1], ToolSearchTool)

@pytest.mark.asyncio

async def test_get_all_tools_keeps_tool_search_for_deferred_hosted_mcp() -> None:

hosted_mcp = HostedMCPTool(

tool_config=cast(

Any,

{

"type": "mcp",

"server_label": "crm_server",

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

"defer_loading": True,

},

)

)

agent = Agent(name="t", tools=[hosted_mcp, ToolSearchTool()])

tools = await agent.get_all_tools(RunContextWrapper(BoolCtx(enable_tools=False)))

assert tools[0] is hosted_mcp

assert isinstance(tools[1], ToolSearchTool)

@pytest.mark.asyncio

async def test_async_failure_error_function_is_awaited() -> None:

async def failure_handler(ctx: RunContextWrapper[Any], exc: Exception) -> str:

return f"handled:{exc}"

@function_tool(failure_error_function=lambda ctx, exc: failure_handler(ctx, exc))

def boom() -> None:

"""Always raises to trigger the failure handler."""

raise RuntimeError("kapow")

ctx = ToolContext(None, tool_name=boom.name, tool_call_id="boom", tool_arguments="{}")

result = await boom.on_invoke_tool(ctx, "{}")

assert result.startswith("handled:")

@pytest.mark.asyncio

async def test_failure_error_function_normalizes_cancelled_error_to_exception() -> None:

seen_error: Exception | None = None

def failure_handler(_ctx: RunContextWrapper[Any], error: Exception) -> str:

nonlocal seen_error

assert isinstance(error, Exception)

assert not isinstance(error, asyncio.CancelledError)

seen_error = error

return f"handled:{error}"

tool = function_tool(lambda: "ok", failure_error_function=failure_handler)

result = await tool_module.maybe_invoke_function_tool_failure_error_function(

function_tool=tool,

context=RunContextWrapper(None),

error=asyncio.CancelledError(),

)

assert result == "handled:Tool execution cancelled."

assert seen_error is not None

assert str(seen_error) == "Tool execution cancelled."

@pytest.mark.asyncio

async def test_default_failure_error_function_is_resolved_at_invoke_time(

monkeypatch: pytest.MonkeyPatch,

) -> None:

def boom(a: int) -> None:

raise ValueError(f"boom:{a}")

tool = function_tool(boom)

def patched_default(_ctx: RunContextWrapper[Any], error: Exception) -> str:

return f"patched:{error}"

monkeypatch.setattr(tool_module, "default_tool_error_function", patched_default)

ctx = ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments='{"a": 7}')

result = await tool.on_invoke_tool(ctx, '{"a": 7}')

assert result == "patched:boom:7"

@pytest.mark.asyncio

async def test_manual_function_tool_uses_default_failure_error_function() -> None:

async def on_invoke_tool(_ctx: ToolContext[Any], _args: str) -> str:

raise asyncio.CancelledError("manual-tool-cancelled")

manual_tool = FunctionTool(

name="manual_cancel_tool",

description="manual cancel",

params_json_schema={},

on_invoke_tool=on_invoke_tool,

)

result = await tool_module.maybe_invoke_function_tool_failure_error_function(

function_tool=manual_tool,

context=RunContextWrapper(None),

error=asyncio.CancelledError("manual-tool-cancelled"),

)

expected = (

"An error occurred while running the tool. Please try again. Error: manual-tool-cancelled"

)

assert result == expected

assert (

tool_module.resolve_function_tool_failure_error_function(manual_tool)

is default_tool_error_function

)

@pytest.mark.asyncio

async def test_failure_error_function_survives_dataclasses_replace() -> None:

def failure_handler(_ctx: RunContextWrapper[Any], error: Exception) -> str:

return f"handled:{error}"

tool = function_tool(lambda: "ok", failure_error_function=failure_handler)

copied_tool = dataclasses.replace(tool, name="copied_tool")

result = await tool_module.maybe_invoke_function_tool_failure_error_function(

function_tool=copied_tool,

context=RunContextWrapper(None),

error=asyncio.CancelledError(),

)

assert result == "handled:Tool execution cancelled."

assert tool_module.resolve_function_tool_failure_error_function(copied_tool) is failure_handler

@pytest.mark.asyncio

async def test_replaced_function_tool_normal_failure_uses_replaced_policy() -> None:

def boom() -> None:

raise RuntimeError("kapow")

replaced_tool = dataclasses.replace(

function_tool(boom),

name="replaced_tool",

_failure_error_function=None,

_use_default_failure_error_function=False,

)

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

await replaced_tool.on_invoke_tool(

ToolContext(None, tool_name=replaced_tool.name, tool_call_id="1", tool_arguments=""),

"",

)

@pytest.mark.asyncio

async def test_shallow_copied_function_tool_normal_failure_uses_copied_policy() -> None:

def boom() -> None:

raise RuntimeError("kapow")

original_tool = function_tool(boom)

custom_state = {"cache": ["alpha"]}

cast(Any, original_tool).custom_state = custom_state

copied_tool = copy.copy(original_tool)

copied_tool.name = "copied_tool"

copied_tool._failure_error_function = None

copied_tool._use_default_failure_error_function = False

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

await copied_tool.on_invoke_tool(

ToolContext(None, tool_name=copied_tool.name, tool_call_id="1", tool_arguments=""),

"",

)

assert cast(Any, copied_tool).custom_state is custom_state

@pytest.mark.asyncio

@pytest.mark.parametrize("copy_style", ["replace", "shallow_copy"])

async def test_copied_function_tool_invalid_input_uses_current_name(copy_style: str) -> None:

def echo(value: str) -> str:

return value

original_tool = function_tool(

echo,

name_override="original_tool",

failure_error_function=None,

)

if copy_style == "replace":

copied_tool = dataclasses.replace(original_tool, name="copied_tool")

else:

copied_tool = copy.copy(original_tool)

copied_tool.name = "copied_tool"

with pytest.raises(ModelBehaviorError, match="Invalid JSON input for tool copied_tool"):

await copied_tool.on_invoke_tool(

ToolContext(

None,

tool_name=copied_tool.name,

tool_call_id="1",

tool_arguments="{}",

),

"{}",

)

def test_function_tool_does_not_mutate_params_json_schema() -> None:

async def noop(ctx: ToolContext[Any], input: str) -> str:

return ""

schema = {"type": "object", "properties": {"x": {"type": "string"}}}

schema_snapshot = copy.deepcopy(schema)

tool = FunctionTool(

name="t",

description="d",

params_json_schema=schema,

on_invoke_tool=noop,

strict_json_schema=True,

)

assert schema == schema_snapshot

assert tool.params_json_schema is not schema

assert tool.params_json_schema["additionalProperties"] is False

assert tool.params_json_schema["required"] == ["x"]

@pytest.mark.asyncio

@pytest.mark.parametrize("input_json", ["[]", '"value"', "123", "null", "true"])

async def test_function_tool_rejects_non_object_json_input(input_json: str) -> None:

def echo(value: str) -> str:

return value

tool = function_tool(

echo,

name_override="echo_tool",

failure_error_function=None,

)

with pytest.raises(

ModelBehaviorError,

match="Invalid JSON input for tool echo_tool: expected a JSON object",

):

await tool.on_invoke_tool(

ToolContext(

None,

tool_name="echo_tool",

tool_call_id="1",

tool_arguments=input_json,

),

input_json,

)

@pytest.mark.asyncio

async def test_function_tool_bad_json_redacts_payload_when_dont_log_tool_data(

monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture

) -> None:

caplog.set_level(logging.DEBUG)

monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", True)

def echo(value: str) -> str:

return value

tool = function_tool(echo, name_override="echo_tool", failure_error_function=None)

bad_json = '{"secret":"SECRET_TOKEN_123"'

with pytest.raises(ModelBehaviorError) as exc_info:

await tool.on_invoke_tool(

ToolContext(

None,

tool_name="echo_tool",

tool_call_id="1",

tool_arguments=bad_json,

),

bad_json,

)

assert str(exc_info.value) == "Invalid JSON input for tool echo_tool"

assert exc_info.value.__cause__ is None

assert exc_info.value.__context__ is None

assert "SECRET_TOKEN_123" not in str(exc_info.value)

assert "SECRET_TOKEN_123" not in caplog.text

@pytest.mark.asyncio

async def test_function_tool_bad_json_includes_payload_when_tool_logging_enabled(

monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture

) -> None:

caplog.set_level(logging.DEBUG)

monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)

def echo(value: str) -> str:

return value

tool = function_tool(echo, name_override="echo_tool", failure_error_function=None)

bad_json = '{"secret":"SECRET_TOKEN_123"'

with pytest.raises(ModelBehaviorError) as exc_info:

await tool.on_invoke_tool(

ToolContext(

None,

tool_name="echo_tool",

tool_call_id="1",

tool_arguments=bad_json,

),

bad_json,

)

assert str(exc_info.value) == f"Invalid JSON input for tool echo_tool: {bad_json}"

assert isinstance(exc_info.value.__cause__, json.JSONDecodeError)

assert exc_info.value.__cause__.doc == bad_json

assert "SECRET_TOKEN_123" in str(exc_info.value)

assert "SECRET_TOKEN_123" in caplog.text

@pytest.mark.asyncio

async def test_function_tool_argument_logging_excludes_live_context(

monkeypatch: pytest.MonkeyPatch,

caplog: pytest.LogCaptureFixture,

) -> None:

context_secret = "CONTEXT_SECRET_SENTINEL"

model_argument = "MODEL_ARGUMENT_SENTINEL"

class SensitiveContext:

def __repr__(self) -> str:

return context_secret

def echo(ctx: ToolContext[Any], value: str) -> str:

assert isinstance(ctx.context, SensitiveContext)

return value

monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)

tool = function_tool(echo)

live_context = ToolContext(

SensitiveContext(),

tool_name=tool.name,

tool_call_id="sensitive-context",

tool_arguments=json.dumps({"value": model_argument}),

)

with caplog.at_level(logging.DEBUG, logger="openai.agents"):

assert (

await tool.on_invoke_tool(

live_context,

json.dumps({"value": model_argument}),

)

== model_argument

)

records = [

record for record in caplog.records if record.msg == "Tool call args: %s, kwargs: %s"

]

assert len(records) == 1

record = records[0]

assert model_argument in logging.Formatter().format(record)

assert context_secret not in repr(record.__dict__)

assert isinstance(record.args, tuple)

logged_args, logged_kwargs = record.args

assert isinstance(logged_args, list)

assert isinstance(logged_kwargs, dict)

assert live_context not in logged_args

assert live_context not in logged_kwargs.values()

@pytest.mark.asyncio

async def test_default_failure_error_function_survives_deepcopy() -> None:

def boom() -> None:

raise RuntimeError("kapow")

tool = function_tool(boom)

copied_tool = copy.deepcopy(tool)

result = await tool_module.maybe_invoke_function_tool_failure_error_function(

function_tool=copied_tool,

context=RunContextWrapper(None),

error=asyncio.CancelledError(),

)

expected = (

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

"Error: Tool execution cancelled."

)

assert result == expected

assert (

tool_module.resolve_function_tool_failure_error_function(copied_tool)

is default_tool_error_function

)

def test_function_tool_accepts_guardrail_arguments():

tool = function_tool(

simple_function,

tool_input_guardrails=[reject_args_guardrail],

tool_output_guardrails=[allow_output_guardrail],

)

assert tool.tool_input_guardrails == [reject_args_guardrail]

assert tool.tool_output_guardrails == [allow_output_guardrail]

def test_function_tool_decorator_accepts_guardrail_arguments():

@function_tool(

tool_input_guardrails=[reject_args_guardrail],

tool_output_guardrails=[allow_output_guardrail],

)

def guarded(a: int) -> int:

return a

assert guarded.tool_input_guardrails == [reject_args_guardrail]

assert guarded.tool_output_guardrails == [allow_output_guardrail]

@pytest.mark.asyncio

async def test_invoke_function_tool_timeout_returns_default_message() -> None:

@function_tool(timeout=0.01)

async def slow_tool() -> str:

await asyncio.sleep(0.2)

return "slow"

ctx = ToolContext(None, tool_name=slow_tool.name, tool_call_id="slow", tool_arguments="{}")

result = await tool_module.invoke_function_tool(

function_tool=slow_tool,

context=ctx,

arguments="{}",

)

assert isinstance(result, str)

assert "timed out" in result.lower()

assert "0.01" in result

@pytest.mark.asyncio

async def test_invoke_function_tool_timeout_uses_custom_error_function() -> None:

def custom_timeout_error(_ctx: RunContextWrapper[Any], error: Exception) -> str:

assert isinstance(error, ToolTimeoutError)

return f"custom_timeout:{error.tool_name}:{error.timeout_seconds:g}"

@function_tool(timeout=0.01, timeout_error_function=custom_timeout_error)

async def slow_tool() -> str:

await asyncio.sleep(0.2)

return "slow"

ctx = ToolContext(None, tool_name=slow_tool.name, tool_call_id="slow", tool_arguments="{}")

result = await tool_module.invoke_function_tool(

function_tool=slow_tool,

context=ctx,

arguments="{}",

)

assert result == "custom_timeout:slow_tool:0.01"

@pytest.mark.asyncio

async def test_invoke_function_tool_timeout_can_raise_exception() -> None:

@function_tool(timeout=0.01, timeout_behavior="raise_exception")

async def slow_tool() -> str:

await asyncio.sleep(0.2)

return "slow"

ctx = ToolContext(None, tool_name=slow_tool.name, tool_call_id="slow", tool_arguments="{}")

with pytest.raises(ToolTimeoutError, match="timed out"):

await tool_module.invoke_function_tool(

function_tool=slow_tool,

context=ctx,

arguments="{}",

)

@pytest.mark.asyncio