Kazuhiro Sera
721 lines
23 KiB
Python
721 lines
23 KiB
Python
import asyncio
|
|
import dataclasses
|
|
import inspect
|
|
import json
|
|
import logging
|
|
from typing import Any, cast
|
|
|
|
import pytest
|
|
from openai.types.responses import ResponseOutputMessage, ResponseOutputText
|
|
from pydantic import BaseModel
|
|
|
|
import agents._debug as _debug
|
|
from agents import (
|
|
Agent,
|
|
Handoff,
|
|
HandoffInputData,
|
|
MessageOutputItem,
|
|
ModelBehaviorError,
|
|
RunContextWrapper,
|
|
UserError,
|
|
handoff,
|
|
)
|
|
from agents._tool_identity import resolve_tool_name_collisions
|
|
from agents.run_internal.run_loop import get_handoffs
|
|
|
|
|
|
def message_item(content: str, agent: Agent[Any]) -> MessageOutputItem:
|
|
return MessageOutputItem(
|
|
agent=agent,
|
|
raw_item=ResponseOutputMessage(
|
|
id="123",
|
|
status="completed",
|
|
role="assistant",
|
|
type="message",
|
|
content=[
|
|
ResponseOutputText(text=content, type="output_text", annotations=[], logprobs=[])
|
|
],
|
|
),
|
|
)
|
|
|
|
|
|
def get_len(data: HandoffInputData) -> int:
|
|
input_len = len(data.input_history) if isinstance(data.input_history, tuple) else 1
|
|
pre_handoff_len = len(data.pre_handoff_items)
|
|
new_items_len = len(data.new_items)
|
|
return input_len + pre_handoff_len + new_items_len
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_single_handoff_setup():
|
|
agent_1 = Agent(name="test_1")
|
|
agent_2 = Agent(name="test_2", handoffs=[agent_1])
|
|
|
|
assert not agent_1.handoffs
|
|
assert agent_2.handoffs == [agent_1]
|
|
|
|
assert not (await get_handoffs(agent_1, RunContextWrapper(agent_1)))
|
|
|
|
handoff_objects = await get_handoffs(agent_2, RunContextWrapper(agent_2))
|
|
assert len(handoff_objects) == 1
|
|
obj = handoff_objects[0]
|
|
assert obj.tool_name == Handoff.default_tool_name(agent_1)
|
|
assert obj.tool_description == Handoff.default_tool_description(agent_1)
|
|
assert obj.agent_name == agent_1.name
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_handoffs_setup():
|
|
agent_1 = Agent(name="test_1")
|
|
agent_2 = Agent(name="test_2")
|
|
agent_3 = Agent(name="test_3", handoffs=[agent_1, agent_2])
|
|
|
|
assert agent_3.handoffs == [agent_1, agent_2]
|
|
assert not agent_1.handoffs
|
|
assert not agent_2.handoffs
|
|
|
|
handoff_objects = await get_handoffs(agent_3, RunContextWrapper(agent_3))
|
|
assert len(handoff_objects) == 2
|
|
assert handoff_objects[0].tool_name == Handoff.default_tool_name(agent_1)
|
|
assert handoff_objects[1].tool_name == Handoff.default_tool_name(agent_2)
|
|
|
|
assert handoff_objects[0].tool_description == Handoff.default_tool_description(agent_1)
|
|
assert handoff_objects[1].tool_description == Handoff.default_tool_description(agent_2)
|
|
|
|
assert handoff_objects[0].agent_name == agent_1.name
|
|
assert handoff_objects[1].agent_name == agent_2.name
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handoffs_reject_colliding_derived_names():
|
|
billing = Agent(name="Billing Agent")
|
|
normalized_billing = Agent(name="billing agent")
|
|
triage = Agent(name="triage", handoffs=[billing, normalized_billing])
|
|
|
|
handoffs = await get_handoffs(triage, RunContextWrapper(None))
|
|
with pytest.raises(UserError) as exc_info:
|
|
resolve_tool_name_collisions((), handoffs, collision_policy="error")
|
|
|
|
assert str(exc_info.value) == (
|
|
"Ambiguous handoff configuration: agents 'Billing Agent' and 'billing agent' both derive "
|
|
"the handoff tool name `transfer_to_billing_agent`. Pass an explicit "
|
|
"`tool_name_override=` to one of them."
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handoff_derived_name_collision_allows_explicit_override():
|
|
billing = Agent(name="Billing Agent")
|
|
normalized_billing = Agent(name="billing agent")
|
|
triage = Agent(
|
|
name="triage",
|
|
handoffs=[
|
|
billing,
|
|
handoff(normalized_billing, tool_name_override="transfer_to_normalized_billing"),
|
|
],
|
|
)
|
|
|
|
handoffs = await get_handoffs(triage, RunContextWrapper(None))
|
|
|
|
assert [item.tool_name for item in handoffs] == [
|
|
"transfer_to_billing_agent",
|
|
"transfer_to_normalized_billing",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handoff_rejects_distinct_agents_with_the_same_name():
|
|
triage = Agent(
|
|
name="triage",
|
|
handoffs=[Agent(name="Billing"), Agent(name="Billing")],
|
|
)
|
|
|
|
handoffs = await get_handoffs(triage, RunContextWrapper(None))
|
|
with pytest.raises(
|
|
UserError,
|
|
match="handoff tool name `transfer_to_billing` is used by multiple handoffs",
|
|
):
|
|
resolve_tool_name_collisions((), handoffs, collision_policy="error")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handoff_warns_and_keeps_last_distinct_agent_with_the_same_name(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
caplog: pytest.LogCaptureFixture,
|
|
):
|
|
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
|
|
triage = Agent(
|
|
name="triage",
|
|
handoffs=[Agent(name="Billing"), Agent(name="Billing")],
|
|
)
|
|
handoffs = await get_handoffs(triage, RunContextWrapper(None))
|
|
|
|
with caplog.at_level("WARNING", logger="openai.agents"):
|
|
_, resolved_handoffs = resolve_tool_name_collisions(
|
|
(),
|
|
handoffs,
|
|
collision_policy="warn",
|
|
)
|
|
|
|
assert resolved_handoffs == [handoffs[1]]
|
|
assert any(
|
|
"handoff tool name `transfer_to_billing` is used by multiple handoffs" in message
|
|
for message in caplog.messages
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handoffs_ignore_disabled_derived_name_collision():
|
|
billing = Agent(name="Billing Agent")
|
|
normalized_billing = Agent(name="billing agent")
|
|
triage = Agent(
|
|
name="triage",
|
|
handoffs=[billing, handoff(normalized_billing, is_enabled=False)],
|
|
)
|
|
|
|
handoffs = await get_handoffs(triage, RunContextWrapper(None))
|
|
|
|
assert [item.agent_name for item in handoffs] == ["Billing Agent"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handoff_default_identity_tracks_the_current_tool_name():
|
|
billing = Agent(name="Billing Agent")
|
|
derived_handoff = handoff(billing)
|
|
copied_handoff = dataclasses.replace(derived_handoff)
|
|
renamed_handoff = dataclasses.replace(
|
|
derived_handoff,
|
|
tool_name="transfer_to_primary_billing",
|
|
)
|
|
normalized_billing = Agent(name="billing agent")
|
|
|
|
copied_triage = Agent(name="copied", handoffs=[copied_handoff, normalized_billing])
|
|
copied_handoffs = await get_handoffs(copied_triage, RunContextWrapper(None))
|
|
with pytest.raises(UserError, match="Ambiguous handoff configuration"):
|
|
resolve_tool_name_collisions((), copied_handoffs, collision_policy="error")
|
|
|
|
renamed_triage = Agent(name="renamed", handoffs=[renamed_handoff, normalized_billing])
|
|
handoffs = await get_handoffs(renamed_triage, RunContextWrapper(None))
|
|
resolve_tool_name_collisions((), handoffs, collision_policy="error")
|
|
assert [item.tool_name for item in handoffs] == [
|
|
"transfer_to_primary_billing",
|
|
"transfer_to_billing_agent",
|
|
]
|
|
|
|
|
|
def test_default_handoff_tool_name_allows_whitespace_without_warning(
|
|
caplog: pytest.LogCaptureFixture,
|
|
):
|
|
agent = Agent(name="Refund agent")
|
|
|
|
with caplog.at_level(logging.WARNING):
|
|
tool_name = Handoff.default_tool_name(agent)
|
|
|
|
assert tool_name == "transfer_to_refund_agent"
|
|
assert not caplog.records
|
|
|
|
|
|
def test_default_handoff_tool_name_warns_for_non_whitespace_invalid_characters(
|
|
caplog: pytest.LogCaptureFixture,
|
|
):
|
|
agent = Agent(name="Refund/agent")
|
|
|
|
with caplog.at_level(logging.WARNING):
|
|
tool_name = Handoff.default_tool_name(agent)
|
|
|
|
assert tool_name == "transfer_to_refund_agent"
|
|
assert len(caplog.records) == 1
|
|
assert "contains invalid characters for function calling" in caplog.records[0].message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_custom_handoff_setup():
|
|
agent_1 = Agent(name="test_1")
|
|
agent_2 = Agent(name="test_2")
|
|
agent_3 = Agent(
|
|
name="test_3",
|
|
handoffs=[
|
|
agent_1,
|
|
handoff(
|
|
agent_2,
|
|
tool_name_override="custom_tool_name",
|
|
tool_description_override="custom tool description",
|
|
),
|
|
],
|
|
)
|
|
|
|
assert len(agent_3.handoffs) == 2
|
|
assert not agent_1.handoffs
|
|
assert not agent_2.handoffs
|
|
|
|
handoff_objects = await get_handoffs(agent_3, RunContextWrapper(agent_3))
|
|
assert len(handoff_objects) == 2
|
|
|
|
first_handoff = handoff_objects[0]
|
|
assert isinstance(first_handoff, Handoff)
|
|
assert first_handoff.tool_name == Handoff.default_tool_name(agent_1)
|
|
assert first_handoff.tool_description == Handoff.default_tool_description(agent_1)
|
|
assert first_handoff.agent_name == agent_1.name
|
|
|
|
second_handoff = handoff_objects[1]
|
|
assert isinstance(second_handoff, Handoff)
|
|
assert second_handoff.tool_name == "custom_tool_name"
|
|
assert second_handoff.tool_description == "custom tool description"
|
|
assert second_handoff.agent_name == agent_2.name
|
|
|
|
|
|
class Foo(BaseModel):
|
|
bar: str
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handoff_input_type():
|
|
async def _on_handoff(ctx: RunContextWrapper[Any], input: Foo):
|
|
pass
|
|
|
|
agent = Agent(name="test")
|
|
obj = handoff(agent, input_type=Foo, on_handoff=_on_handoff)
|
|
for key, value in Foo.model_json_schema().items():
|
|
assert obj.input_json_schema[key] == value
|
|
|
|
# Invalid JSON should raise an error
|
|
with pytest.raises(ModelBehaviorError):
|
|
await obj.on_invoke_handoff(RunContextWrapper(agent), "not json")
|
|
|
|
# Empty JSON should raise an error
|
|
with pytest.raises(ModelBehaviorError):
|
|
await obj.on_invoke_handoff(RunContextWrapper(agent), "")
|
|
|
|
# Valid JSON should call the on_handoff function
|
|
invoked = await obj.on_invoke_handoff(
|
|
RunContextWrapper(agent), Foo(bar="baz").model_dump_json()
|
|
)
|
|
assert invoked == agent
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_handoff_called():
|
|
was_called = False
|
|
|
|
async def _on_handoff(ctx: RunContextWrapper[Any], input: Foo):
|
|
nonlocal was_called
|
|
was_called = True
|
|
|
|
agent = Agent(name="test")
|
|
obj = handoff(agent, input_type=Foo, on_handoff=_on_handoff)
|
|
for key, value in Foo.model_json_schema().items():
|
|
assert obj.input_json_schema[key] == value
|
|
|
|
invoked = await obj.on_invoke_handoff(
|
|
RunContextWrapper(agent), Foo(bar="baz").model_dump_json()
|
|
)
|
|
assert invoked == agent
|
|
|
|
assert was_called, "on_handoff should have been called"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_handoff_without_input_called():
|
|
was_called = False
|
|
|
|
def _on_handoff(ctx: RunContextWrapper[Any]):
|
|
nonlocal was_called
|
|
was_called = True
|
|
|
|
agent = Agent(name="test")
|
|
obj = handoff(agent, on_handoff=_on_handoff)
|
|
|
|
invoked = await obj.on_invoke_handoff(RunContextWrapper(agent), "")
|
|
assert invoked == agent
|
|
|
|
assert was_called, "on_handoff should have been called"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_on_handoff_without_input_called():
|
|
was_called = False
|
|
|
|
async def _on_handoff(ctx: RunContextWrapper[Any]):
|
|
nonlocal was_called
|
|
was_called = True
|
|
|
|
agent = Agent(name="test")
|
|
obj = handoff(agent, on_handoff=_on_handoff)
|
|
|
|
invoked = await obj.on_invoke_handoff(RunContextWrapper(agent), "")
|
|
assert invoked == agent
|
|
|
|
assert was_called, "on_handoff should have been called"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_callable_class_with_async_dunder_call_is_awaited():
|
|
"""Callable instances whose ``__call__`` is async must be awaited.
|
|
|
|
``inspect.iscoroutinefunction`` returns ``False`` for the instance itself, so the
|
|
previous implementation invoked it without awaiting and silently dropped the
|
|
coroutine.
|
|
"""
|
|
|
|
class WithInput:
|
|
def __init__(self) -> None:
|
|
self.calls: list[Foo] = []
|
|
|
|
async def __call__(self, ctx: RunContextWrapper[Any], input: Foo) -> None:
|
|
self.calls.append(input)
|
|
|
|
class NoInput:
|
|
def __init__(self) -> None:
|
|
self.calls = 0
|
|
|
|
async def __call__(self, ctx: RunContextWrapper[Any]) -> None:
|
|
self.calls += 1
|
|
|
|
agent = Agent(name="test")
|
|
|
|
with_input_cb = WithInput()
|
|
obj_with = handoff(agent, input_type=Foo, on_handoff=with_input_cb)
|
|
await obj_with.on_invoke_handoff(RunContextWrapper(agent), Foo(bar="baz").model_dump_json())
|
|
assert with_input_cb.calls == [Foo(bar="baz")]
|
|
|
|
no_input_cb = NoInput()
|
|
obj_no = handoff(agent, on_handoff=no_input_cb)
|
|
await obj_no.on_invoke_handoff(RunContextWrapper(agent), "")
|
|
assert no_input_cb.calls == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_on_handoff_raises_error():
|
|
was_called = False
|
|
|
|
async def _on_handoff(ctx: RunContextWrapper[Any], blah: str):
|
|
nonlocal was_called
|
|
was_called = True # pragma: no cover
|
|
|
|
agent = Agent(name="test")
|
|
|
|
with pytest.raises(UserError):
|
|
# Purposely ignoring the type error here to simulate invalid input
|
|
handoff(agent, on_handoff=_on_handoff) # type: ignore
|
|
|
|
|
|
def test_input_type_without_on_handoff_raises_error():
|
|
"""Providing input_type without on_handoff should raise an error."""
|
|
|
|
class MyInput(BaseModel):
|
|
reason: str
|
|
|
|
agent = Agent(name="test")
|
|
|
|
with pytest.raises(UserError, match="You must provide on_handoff when input_type is provided"):
|
|
handoff(agent, input_type=MyInput) # type: ignore
|
|
|
|
|
|
def test_non_callable_on_handoff_with_input_type_raises_error():
|
|
"""Providing a non-callable on_handoff with input_type should raise an error."""
|
|
|
|
class MyInput(BaseModel):
|
|
reason: str
|
|
|
|
agent = Agent(name="test")
|
|
|
|
with pytest.raises(UserError, match="on_handoff must be callable"):
|
|
handoff(agent, on_handoff="not_a_function", input_type=MyInput) # type: ignore
|
|
|
|
|
|
def test_handoff_input_data():
|
|
agent = Agent(name="test")
|
|
|
|
data = HandoffInputData(
|
|
input_history="",
|
|
pre_handoff_items=(),
|
|
new_items=(),
|
|
run_context=RunContextWrapper(context=()),
|
|
)
|
|
assert get_len(data) == 1
|
|
|
|
data = HandoffInputData(
|
|
input_history=({"role": "user", "content": "foo"},),
|
|
pre_handoff_items=(),
|
|
new_items=(),
|
|
run_context=RunContextWrapper(context=()),
|
|
)
|
|
assert get_len(data) == 1
|
|
|
|
data = HandoffInputData(
|
|
input_history=(
|
|
{"role": "user", "content": "foo"},
|
|
{"role": "assistant", "content": "bar"},
|
|
),
|
|
pre_handoff_items=(),
|
|
new_items=(),
|
|
run_context=RunContextWrapper(context=()),
|
|
)
|
|
assert get_len(data) == 2
|
|
|
|
data = HandoffInputData(
|
|
input_history=({"role": "user", "content": "foo"},),
|
|
pre_handoff_items=(
|
|
message_item("foo", agent),
|
|
message_item("foo2", agent),
|
|
),
|
|
new_items=(
|
|
message_item("bar", agent),
|
|
message_item("baz", agent),
|
|
),
|
|
run_context=RunContextWrapper(context=()),
|
|
)
|
|
assert get_len(data) == 5
|
|
|
|
data = HandoffInputData(
|
|
input_history=(
|
|
{"role": "user", "content": "foo"},
|
|
{"role": "assistant", "content": "bar"},
|
|
),
|
|
pre_handoff_items=(message_item("baz", agent),),
|
|
new_items=(
|
|
message_item("baz", agent),
|
|
message_item("qux", agent),
|
|
),
|
|
run_context=RunContextWrapper(context=()),
|
|
)
|
|
|
|
assert get_len(data) == 5
|
|
|
|
|
|
def test_handoff_input_schema_is_strict():
|
|
agent = Agent(name="test")
|
|
obj = handoff(agent, input_type=Foo, on_handoff=lambda ctx, input: None)
|
|
for key, value in Foo.model_json_schema().items():
|
|
assert obj.input_json_schema[key] == value
|
|
|
|
assert obj.strict_json_schema, "Input schema should be strict"
|
|
|
|
assert (
|
|
"additionalProperties" in obj.input_json_schema
|
|
and not obj.input_json_schema["additionalProperties"]
|
|
), "Input schema should be strict and have additionalProperties=False"
|
|
|
|
|
|
def test_handoff_rejects_nullable_input_root():
|
|
agent = Agent(name="test")
|
|
|
|
with pytest.raises(UserError, match="root of a strict JSON schema"):
|
|
handoff(
|
|
agent,
|
|
input_type=cast(type[Any], Foo | None),
|
|
on_handoff=lambda ctx, input: None,
|
|
)
|
|
|
|
|
|
def test_get_transfer_message_is_valid_json() -> None:
|
|
agent = Agent(name="foo")
|
|
obj = handoff(agent)
|
|
transfer = obj.get_transfer_message(agent)
|
|
assert json.loads(transfer) == {"assistant": agent.name}
|
|
|
|
|
|
def test_handoff_is_enabled_bool():
|
|
"""Test that handoff respects is_enabled boolean parameter."""
|
|
agent = Agent(name="test")
|
|
|
|
# Test enabled handoff (default)
|
|
handoff_enabled = handoff(agent)
|
|
assert handoff_enabled.is_enabled is True
|
|
|
|
# Test explicitly enabled handoff
|
|
handoff_explicit_enabled = handoff(agent, is_enabled=True)
|
|
assert handoff_explicit_enabled.is_enabled is True
|
|
|
|
# Test disabled handoff
|
|
handoff_disabled = handoff(agent, is_enabled=False)
|
|
assert handoff_disabled.is_enabled is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handoff_is_enabled_callable():
|
|
"""Test that handoff respects is_enabled callable parameter."""
|
|
agent = Agent(name="test")
|
|
|
|
# Test callable that returns True
|
|
def always_enabled(ctx: RunContextWrapper[Any], agent: Agent[Any]) -> bool:
|
|
return True
|
|
|
|
handoff_callable_enabled = handoff(agent, is_enabled=always_enabled)
|
|
assert callable(handoff_callable_enabled.is_enabled)
|
|
result = handoff_callable_enabled.is_enabled(RunContextWrapper(agent), agent)
|
|
assert inspect.isawaitable(result)
|
|
result = await result
|
|
assert result is True
|
|
|
|
# Test callable that returns False
|
|
def always_disabled(ctx: RunContextWrapper[Any], agent: Agent[Any]) -> bool:
|
|
return False
|
|
|
|
handoff_callable_disabled = handoff(agent, is_enabled=always_disabled)
|
|
assert callable(handoff_callable_disabled.is_enabled)
|
|
result = handoff_callable_disabled.is_enabled(RunContextWrapper(agent), agent)
|
|
assert inspect.isawaitable(result)
|
|
result = await result
|
|
assert result is False
|
|
|
|
# Test async callable
|
|
async def async_enabled(ctx: RunContextWrapper[Any], agent: Agent[Any]) -> bool:
|
|
return True
|
|
|
|
handoff_async_enabled = handoff(agent, is_enabled=async_enabled)
|
|
assert callable(handoff_async_enabled.is_enabled)
|
|
result = await handoff_async_enabled.is_enabled(RunContextWrapper(agent), agent) # type: ignore
|
|
assert result is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handoff_is_enabled_filtering_integration():
|
|
"""Integration test that disabled handoffs are filtered out by the runner."""
|
|
|
|
# Set up agents
|
|
agent_1 = Agent(name="agent_1")
|
|
agent_2 = Agent(name="agent_2")
|
|
agent_3 = Agent(name="agent_3")
|
|
|
|
# Create main agent with mixed enabled/disabled handoffs
|
|
main_agent = Agent(
|
|
name="main_agent",
|
|
handoffs=[
|
|
handoff(agent_1, is_enabled=True), # enabled
|
|
handoff(agent_2, is_enabled=False), # disabled
|
|
handoff(agent_3, is_enabled=lambda ctx, agent: True), # enabled callable
|
|
],
|
|
)
|
|
|
|
context_wrapper = RunContextWrapper(main_agent)
|
|
|
|
# Get filtered handoffs using the runner's method
|
|
filtered_handoffs = await get_handoffs(main_agent, context_wrapper)
|
|
|
|
# Should only have 2 handoffs (agent_1 and agent_3), agent_2 should be filtered out
|
|
assert len(filtered_handoffs) == 2
|
|
|
|
# Check that the correct agents are present
|
|
agent_names = {h.agent_name for h in filtered_handoffs}
|
|
assert agent_names == {"agent_1", "agent_3"}
|
|
assert "agent_2" not in agent_names
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_handoffs_cancels_sibling_enablement_checks_on_error() -> None:
|
|
slow_started = asyncio.Event()
|
|
slow_cancelled = asyncio.Event()
|
|
slow_finished = asyncio.Event()
|
|
|
|
async def slow_enabled(_ctx: RunContextWrapper[Any], _agent: Agent[Any]) -> bool:
|
|
slow_started.set()
|
|
try:
|
|
await asyncio.Event().wait()
|
|
except asyncio.CancelledError:
|
|
slow_cancelled.set()
|
|
raise
|
|
finally:
|
|
slow_finished.set()
|
|
return True
|
|
|
|
async def failing_enabled(_ctx: RunContextWrapper[Any], _agent: Agent[Any]) -> bool:
|
|
await slow_started.wait()
|
|
raise RuntimeError("enablement failed")
|
|
|
|
parent = Agent(
|
|
name="parent",
|
|
handoffs=[
|
|
handoff(Agent(name="slow"), is_enabled=slow_enabled),
|
|
handoff(Agent(name="failing"), is_enabled=failing_enabled),
|
|
],
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="enablement failed"):
|
|
await get_handoffs(parent, RunContextWrapper(None))
|
|
|
|
assert slow_cancelled.is_set()
|
|
assert slow_finished.is_set()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handoff_is_enabled_sync_callable_false_filters_handoff():
|
|
target_agent = Agent(name="target")
|
|
main_agent = Agent(
|
|
name="main",
|
|
handoffs=[handoff(target_agent, is_enabled=lambda ctx, agent: False)],
|
|
)
|
|
|
|
filtered_handoffs = await get_handoffs(main_agent, RunContextWrapper(main_agent))
|
|
|
|
assert filtered_handoffs == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handoff_direct_sync_is_enabled_callable_filters_handoff():
|
|
async def invoke_handoff(ctx: RunContextWrapper[Any], input_json: str) -> Agent[Any]:
|
|
_ = (ctx, input_json)
|
|
return Agent(name="target")
|
|
|
|
handoff_obj = Handoff(
|
|
tool_name="transfer_to_target",
|
|
tool_description="Transfer to target.",
|
|
input_json_schema={},
|
|
on_invoke_handoff=invoke_handoff,
|
|
agent_name="target",
|
|
is_enabled=lambda ctx, agent: False,
|
|
)
|
|
main_agent = Agent(name="main", handoffs=[handoff_obj])
|
|
|
|
filtered_handoffs = await get_handoffs(main_agent, RunContextWrapper(main_agent))
|
|
|
|
assert filtered_handoffs == []
|
|
|
|
|
|
class StrictInput(BaseModel):
|
|
name: str
|
|
age: int
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handoff_strict_json_rejects_type_coercion():
|
|
"""With strict_json_schema=True (default), string input for an int field must raise
|
|
ModelBehaviorError instead of being silently coerced."""
|
|
|
|
async def _on_handoff(ctx: RunContextWrapper[Any], input: StrictInput):
|
|
pass # pragma: no cover
|
|
|
|
agent = Agent(name="test")
|
|
obj = handoff(agent, input_type=StrictInput, on_handoff=_on_handoff)
|
|
|
|
# strict_json_schema defaults to True
|
|
assert obj.strict_json_schema is True
|
|
|
|
# age is a string "25" — strict mode should reject this
|
|
malformed_json = '{"name": "Alice", "age": "25"}'
|
|
with pytest.raises(ModelBehaviorError, match="Invalid JSON"):
|
|
await obj.on_invoke_handoff(RunContextWrapper(agent), malformed_json)
|
|
|
|
# Correctly typed input should still be accepted
|
|
valid_json = '{"name": "Alice", "age": 25}'
|
|
result = await obj.on_invoke_handoff(RunContextWrapper(agent), valid_json)
|
|
assert result == agent
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handoff_lenient_json_allows_type_coercion():
|
|
"""Without strict validation, Pydantic's default lenient mode silently coerces
|
|
string input for an int field — verifying backward compatibility."""
|
|
from pydantic import TypeAdapter
|
|
|
|
from agents.util._json import validate_json
|
|
|
|
type_adapter = TypeAdapter(StrictInput)
|
|
|
|
# age is a string "25" — lenient mode should coerce it to int 25
|
|
malformed_json = '{"name": "Alice", "age": "25"}'
|
|
result = validate_json(malformed_json, type_adapter, partial=False)
|
|
assert result.name == "Alice"
|
|
assert result.age == 25
|
|
assert isinstance(result.age, int)
|