diff --git a/docs/tools.md b/docs/tools.md index 17f7da0a1..1684b10c0 100644 --- a/docs/tools.md +++ b/docs/tools.md @@ -311,6 +311,74 @@ json_tool = data_agent.as_tool( ) ``` +### Conditional tool enabling + +You can conditionally enable or disable agent tools at runtime using the `is_enabled` parameter. This allows you to dynamically filter which tools are available to the LLM based on context, user preferences, or runtime conditions. + +```python +import asyncio +from agents import Agent, AgentBase, Runner, RunContextWrapper +from pydantic import BaseModel + +class LanguageContext(BaseModel): + language_preference: str = "french_spanish" + +def french_enabled(ctx: RunContextWrapper[LanguageContext], agent: AgentBase) -> bool: + """Enable French for French+Spanish preference.""" + return ctx.context.language_preference == "french_spanish" + +# Create specialized agents +spanish_agent = Agent( + name="spanish_agent", + instructions="You respond in Spanish. Always reply to the user's question in Spanish.", +) + +french_agent = Agent( + name="french_agent", + instructions="You respond in French. Always reply to the user's question in French.", +) + +# Create orchestrator with conditional tools +orchestrator = Agent( + name="orchestrator", + instructions=( + "You are a multilingual assistant. You use the tools given to you to respond to users. " + "You must call ALL available tools to provide responses in different languages. " + "You never respond in languages yourself, you always use the provided tools." + ), + tools=[ + spanish_agent.as_tool( + tool_name="respond_spanish", + tool_description="Respond to the user's question in Spanish", + is_enabled=True, # Always enabled + ), + french_agent.as_tool( + tool_name="respond_french", + tool_description="Respond to the user's question in French", + is_enabled=french_enabled, + ), + ], +) + +async def main(): + context = RunContextWrapper(LanguageContext(language_preference="french_spanish")) + result = await Runner.run(orchestrator, "How are you?", context=context.context) + print(result.final_output) + +asyncio.run(main()) +``` + +The `is_enabled` parameter accepts: +- **Boolean values**: `True` (always enabled) or `False` (always disabled) +- **Callable functions**: Functions that take `(context, agent)` and return a boolean +- **Async functions**: Async functions for complex conditional logic + +Disabled tools are completely hidden from the LLM at runtime, making this useful for: +- Feature gating based on user permissions +- Environment-specific tool availability (dev vs prod) +- A/B testing different tool configurations +- Dynamic tool filtering based on runtime state + ## Handling errors in function tools When you create a function tool via `@function_tool`, you can pass a `failure_error_function`. This is a function that provides an error response to the LLM in case the tool call crashes. diff --git a/examples/agent_patterns/agents_as_tools_conditional.py b/examples/agent_patterns/agents_as_tools_conditional.py new file mode 100644 index 000000000..e00f56d5e --- /dev/null +++ b/examples/agent_patterns/agents_as_tools_conditional.py @@ -0,0 +1,113 @@ +import asyncio + +from pydantic import BaseModel + +from agents import Agent, AgentBase, RunContextWrapper, Runner, trace + +""" +This example demonstrates the agents-as-tools pattern with conditional tool enabling. +Agent tools are dynamically enabled/disabled based on user access levels using the +is_enabled parameter. +""" + + +class AppContext(BaseModel): + language_preference: str = "spanish_only" # "spanish_only", "french_spanish", "european" + + +def french_spanish_enabled(ctx: RunContextWrapper[AppContext], agent: AgentBase) -> bool: + """Enable for French+Spanish and European preferences.""" + return ctx.context.language_preference in ["french_spanish", "european"] + + +def european_enabled(ctx: RunContextWrapper[AppContext], agent: AgentBase) -> bool: + """Only enable for European preference.""" + return ctx.context.language_preference == "european" + + +# Create specialized agents +spanish_agent = Agent( + name="spanish_agent", + instructions="You respond in Spanish. Always reply to the user's question in Spanish.", +) + +french_agent = Agent( + name="french_agent", + instructions="You respond in French. Always reply to the user's question in French.", +) + +italian_agent = Agent( + name="italian_agent", + instructions="You respond in Italian. Always reply to the user's question in Italian.", +) + +# Create orchestrator with conditional tools +orchestrator = Agent( + name="orchestrator", + instructions=( + "You are a multilingual assistant. You use the tools given to you to respond to users. " + "You must call ALL available tools to provide responses in different languages. " + "You never respond in languages yourself, you always use the provided tools." + ), + tools=[ + spanish_agent.as_tool( + tool_name="respond_spanish", + tool_description="Respond to the user's question in Spanish", + is_enabled=True, # Always enabled + ), + french_agent.as_tool( + tool_name="respond_french", + tool_description="Respond to the user's question in French", + is_enabled=french_spanish_enabled, + ), + italian_agent.as_tool( + tool_name="respond_italian", + tool_description="Respond to the user's question in Italian", + is_enabled=european_enabled, + ), + ], +) + + +async def main(): + """Interactive demo with LLM interaction.""" + print("Agents-as-Tools with Conditional Enabling\n") + print( + "This demonstrates how language response tools are dynamically enabled based on user preferences.\n" + ) + + print("Choose language preference:") + print("1. Spanish only (1 tool)") + print("2. French and Spanish (2 tools)") + print("3. European languages (3 tools)") + + choice = input("\nSelect option (1-3): ").strip() + preference_map = {"1": "spanish_only", "2": "french_spanish", "3": "european"} + language_preference = preference_map.get(choice, "spanish_only") + + # Create context and show available tools + context = RunContextWrapper(AppContext(language_preference=language_preference)) + available_tools = await orchestrator.get_all_tools(context) + tool_names = [tool.name for tool in available_tools] + + print(f"\nLanguage preference: {language_preference}") + print(f"Available tools: {', '.join(tool_names)}") + print(f"The LLM will only see and can use these {len(available_tools)} tools\n") + + # Get user request + user_request = input("Ask a question and see responses in available languages:\n") + + # Run with LLM interaction + print("\nProcessing request...") + with trace("Conditional tool access"): + result = await Runner.run( + starting_agent=orchestrator, + input=user_request, + context=context.context, + ) + + print(f"\nResponse:\n{result.final_output}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/agents/agent.py b/src/agents/agent.py index b67a12c0d..4982da041 100644 --- a/src/agents/agent.py +++ b/src/agents/agent.py @@ -234,6 +234,8 @@ def as_tool( tool_name: str | None, tool_description: str | None, custom_output_extractor: Callable[[RunResult], Awaitable[str]] | None = None, + is_enabled: bool + | Callable[[RunContextWrapper[Any], AgentBase[Any]], MaybeAwaitable[bool]] = True, ) -> Tool: """Transform this agent into a tool, callable by other agents. @@ -249,11 +251,15 @@ def as_tool( when to use it. custom_output_extractor: A function that extracts the output from the agent. If not provided, the last message from the agent will be used. + is_enabled: Whether the tool is enabled. Can be a bool or a callable that takes the run + context and agent and returns whether the tool is enabled. Disabled tools are hidden + from the LLM at runtime. """ @function_tool( name_override=tool_name or _transforms.transform_string_function_style(self.name), description_override=tool_description or "", + is_enabled=is_enabled, ) async def run_agent(context: RunContextWrapper, input: str) -> str: from .run import Runner diff --git a/tests/test_agent_as_tool.py b/tests/test_agent_as_tool.py new file mode 100644 index 000000000..3307c7a1a --- /dev/null +++ b/tests/test_agent_as_tool.py @@ -0,0 +1,207 @@ +import pytest +from pydantic import BaseModel + +from agents import Agent, AgentBase, FunctionTool, RunContextWrapper + + +class BoolCtx(BaseModel): + enable_tools: bool + + +@pytest.mark.asyncio +async def test_agent_as_tool_is_enabled_bool(): + """Test that agent.as_tool() respects static boolean is_enabled parameter.""" + # Create a simple agent + agent = Agent( + name="test_agent", + instructions="You are a test agent that says hello.", + ) + + # Create tool with is_enabled=False + disabled_tool = agent.as_tool( + tool_name="disabled_agent_tool", + tool_description="A disabled agent tool", + is_enabled=False, + ) + + # Create tool with is_enabled=True (default) + enabled_tool = agent.as_tool( + tool_name="enabled_agent_tool", + tool_description="An enabled agent tool", + is_enabled=True, + ) + + # Create another tool with default is_enabled (should be True) + default_tool = agent.as_tool( + tool_name="default_agent_tool", + tool_description="A default agent tool", + ) + + # Create test agent that uses these tools + orchestrator = Agent( + name="orchestrator", + instructions="You orchestrate other agents.", + tools=[disabled_tool, enabled_tool, default_tool], + ) + + # Test with any context + context = RunContextWrapper(BoolCtx(enable_tools=True)) + + # Get all tools - should filter out the disabled one + tools = await orchestrator.get_all_tools(context) + tool_names = [tool.name for tool in tools] + + assert "enabled_agent_tool" in tool_names + assert "default_agent_tool" in tool_names + assert "disabled_agent_tool" not in tool_names + + +@pytest.mark.asyncio +async def test_agent_as_tool_is_enabled_callable(): + """Test that agent.as_tool() respects callable is_enabled parameter.""" + # Create a simple agent + agent = Agent( + name="test_agent", + instructions="You are a test agent that says hello.", + ) + + # Create tool with callable is_enabled + async def cond_enabled(ctx: RunContextWrapper[BoolCtx], agent: AgentBase) -> bool: + return ctx.context.enable_tools + + conditional_tool = agent.as_tool( + tool_name="conditional_agent_tool", + tool_description="A conditionally enabled agent tool", + is_enabled=cond_enabled, + ) + + # Create tool with lambda is_enabled + lambda_tool = agent.as_tool( + tool_name="lambda_agent_tool", + tool_description="A lambda enabled agent tool", + is_enabled=lambda ctx, agent: ctx.context.enable_tools, + ) + + # Create test agent that uses these tools + orchestrator = Agent( + name="orchestrator", + instructions="You orchestrate other agents.", + tools=[conditional_tool, lambda_tool], + ) + + # Test with enable_tools=False + context_disabled = RunContextWrapper(BoolCtx(enable_tools=False)) + tools_disabled = await orchestrator.get_all_tools(context_disabled) + assert len(tools_disabled) == 0 + + # Test with enable_tools=True + context_enabled = RunContextWrapper(BoolCtx(enable_tools=True)) + tools_enabled = await orchestrator.get_all_tools(context_enabled) + tool_names = [tool.name for tool in tools_enabled] + + assert len(tools_enabled) == 2 + assert "conditional_agent_tool" in tool_names + assert "lambda_agent_tool" in tool_names + + +@pytest.mark.asyncio +async def test_agent_as_tool_is_enabled_mixed(): + """Test agent.as_tool() with mixed enabled/disabled tools.""" + # Create a simple agent + agent = Agent( + name="test_agent", + instructions="You are a test agent that says hello.", + ) + + # Create various tools with different is_enabled configurations + always_enabled = agent.as_tool( + tool_name="always_enabled", + tool_description="Always enabled tool", + is_enabled=True, + ) + + always_disabled = agent.as_tool( + tool_name="always_disabled", + tool_description="Always disabled tool", + is_enabled=False, + ) + + conditionally_enabled = agent.as_tool( + tool_name="conditionally_enabled", + tool_description="Conditionally enabled tool", + is_enabled=lambda ctx, agent: ctx.context.enable_tools, + ) + + default_enabled = agent.as_tool( + tool_name="default_enabled", + tool_description="Default enabled tool", + ) + + # Create test agent that uses these tools + orchestrator = Agent( + name="orchestrator", + instructions="You orchestrate other agents.", + tools=[always_enabled, always_disabled, conditionally_enabled, default_enabled], + ) + + # Test with enable_tools=False + context_disabled = RunContextWrapper(BoolCtx(enable_tools=False)) + tools_disabled = await orchestrator.get_all_tools(context_disabled) + tool_names_disabled = [tool.name for tool in tools_disabled] + + assert len(tools_disabled) == 2 + assert "always_enabled" in tool_names_disabled + assert "default_enabled" in tool_names_disabled + assert "always_disabled" not in tool_names_disabled + assert "conditionally_enabled" not in tool_names_disabled + + # Test with enable_tools=True + context_enabled = RunContextWrapper(BoolCtx(enable_tools=True)) + tools_enabled = await orchestrator.get_all_tools(context_enabled) + tool_names_enabled = [tool.name for tool in tools_enabled] + + assert len(tools_enabled) == 3 + assert "always_enabled" in tool_names_enabled + assert "default_enabled" in tool_names_enabled + assert "conditionally_enabled" in tool_names_enabled + assert "always_disabled" not in tool_names_enabled + + +@pytest.mark.asyncio +async def test_agent_as_tool_is_enabled_preserves_other_params(): + """Test that is_enabled parameter doesn't interfere with other agent.as_tool() parameters.""" + # Create a simple agent + agent = Agent( + name="test_agent", + instructions="You are a test agent that returns a greeting.", + ) + + # Custom output extractor + async def custom_extractor(result): + return f"CUSTOM: {result.new_items[-1].text if result.new_items else 'No output'}" + + # Create tool with all parameters including is_enabled + tool = agent.as_tool( + tool_name="custom_tool_name", + tool_description="A custom tool with all parameters", + custom_output_extractor=custom_extractor, + is_enabled=True, + ) + + # Verify the tool was created with correct properties + assert tool.name == "custom_tool_name" + assert isinstance(tool, FunctionTool) + assert tool.description == "A custom tool with all parameters" + assert tool.is_enabled is True + + # Verify tool is included when enabled + orchestrator = Agent( + name="orchestrator", + instructions="You orchestrate other agents.", + tools=[tool], + ) + + context = RunContextWrapper(BoolCtx(enable_tools=True)) + tools = await orchestrator.get_all_tools(context) + assert len(tools) == 1 + assert tools[0].name == "custom_tool_name"