Loading...
Loading...
Test AG2 agents and tools without hitting a real LLM provider. Pass `TestConfig(...)` from `ag2.testing` as the agent's config (or per-`ask`) to mock LLM responses, inject `ToolCallEvent`s to simulate tool execution, and assert success / error paths. Use when the user is writing pytest tests for an Agent or Tool.
npx skill4agent add ag2ai/ag2-skills ag2-testingAgent@toolimport pytest
from ag2 import Agent
from ag2.testing import TestConfig
@pytest.mark.asyncio
async def test_mocked_response():
agent = Agent("test_agent")
reply = await agent.ask("Hi!", config=TestConfig("This is a mocked response."))
assert reply.body == "This is a mocked response."TestConfig(*responses)ask()ToolCallEventask()ToolCallEventimport pytest
from ag2 import Agent
from ag2.events import ToolCallEvent
from ag2.testing import TestConfig
@pytest.mark.asyncio
async def test_tool_success():
def my_tool() -> str:
return "tool execution result"
agent = Agent("test_agent", tools=[my_tool])
config = TestConfig(
ToolCallEvent(name="my_tool"),
"final result",
)
reply = await agent.ask("Please use my_tool", config=config)
assert reply.body == "final result"ask()@pytest.mark.asyncio
async def test_tool_raises():
def failing_tool() -> str:
raise ValueError("Something went wrong")
config = TestConfig(
ToolCallEvent(name="failing_tool"),
"result",
)
agent = Agent("test_agent", config=config, tools=[failing_tool])
with pytest.raises(ValueError, match="Something went wrong"):
await agent.ask("Hi!")ToolNotFoundErrorfrom ag2.exceptions import ToolNotFoundError
@pytest.mark.asyncio
async def test_tool_not_found():
config = TestConfig(ToolCallEvent(name="unregistered_tool"))
agent = Agent("test_agent", config=config)
with pytest.raises(ToolNotFoundError, match="Tool `unregistered_tool` not found"):
await agent.ask("Hi!")Dependsdef get_production_db():
raise Exception("Do not call in tests!")
@tool
def read_data(db: Annotated[object, Depends(get_production_db)]) -> str:
return "Data"
agent = Agent("test", tools=[read_data])
agent.dependency_provider.override(get_production_db, lambda: "mock_db")Injectdependencies={...}agent.ask(...)await agent.ask("Read", dependencies={"database_pool": fake_pool})from ag2 import MemoryStream
from ag2.events import ToolCallEvent
stream = MemoryStream()
collected: list[ToolCallEvent] = []
stream.where(ToolCallEvent).subscribe(lambda e: collected.append(e))
await agent.ask("Test", stream=stream)
assert collected[0].name == "expected_tool"ask()TestConfig(...)ask()TestConfig(ToolCallEvent("my_tool"), "final result")"final result"TestConfig.create()responses[0]ask()ask()reply.ask(...)ask()TestConfig(...)ask()config=await agent.ask("…", config=TestConfig("turn-2 reply"))ToolCallEventworkflowdiscussionconfig=website/docs/user-guide/testing.mdxpyproject.toml@pytest.mark.asynciowebsite/docs/user-guide/advanced/stream.mdx@pytest.mark.asyncioTestConfigask()StopIterationask()Dependsreply.bodyresponse_schemabodyawait reply.content()AgentTestConfigOPENAI_API_KEY