yanis/autogen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update OpenAIAgent to reflect gap in supporting custom function tool (#6943)

Browse Source
This commit is contained in:
Eric Zhu
2025-08-19 00:23:37 -07:00
committed by GitHub
parent fbc7888d75
commit d4dd4a26ca
3 changed files with 93 additions and 763 deletions
Download Patch File Download Diff File Expand all files Collapse all files

561
python/packages/autogen-ext/src/autogen_ext/agents/openai/_openai_agent.py
View File

@@ -1,7 +1,5 @@
import asyncio
import json
import logging
import warnings
from typing import (
Any,
AsyncGenerator,
@@ -30,14 +28,12 @@ from autogen_agentchat.messages import (
TextMessage,
ToolCallSummaryMessage,
)
from autogen_core import CancellationToken, Component, ComponentModel, FunctionCall
from autogen_core import CancellationToken, Component
from autogen_core.models import UserMessage
from autogen_core.tools import Tool
from pydantic import BaseModel, Field
from typing_extensions import NotRequired, TypedDict
from openai import AsyncAzureOpenAI, AsyncOpenAI # type: ignore
from openai.types.responses import FunctionToolParam
# Number of characters to display when previewing image content in logs and UI
# Base64 encoded images can be very long, so we truncate for readability
@@ -48,29 +44,6 @@ IMAGE_CONTENT_PREVIEW_LENGTH = 50
event_logger = logging.getLogger(EVENT_LOGGER_NAME)
def _convert_tool_to_function_tool_param(tool: Tool) -> FunctionToolParam:
"""Convert an autogen Tool to an OpenAI Responses function tool parameter."""
schema = tool.schema
parameters: Dict[str, object] = {}
if "parameters" in schema:
parameters = {
"type": schema["parameters"]["type"],
"properties": schema["parameters"]["properties"],
}
if "required" in schema["parameters"]:
parameters["required"] = schema["parameters"]["required"]
function_tool_param = FunctionToolParam(
type="function",
name=schema["name"],
description=schema.get("description", ""),
parameters=parameters,
strict=schema.get("strict", None),
)
return function_tool_param
# TypedDict classes for built-in tool configurations
class FileSearchToolConfig(TypedDict):
"""Configuration for file_search tool."""
@@ -86,7 +59,7 @@ class WebSearchToolConfig(TypedDict):
"""Configuration for web_search_preview tool."""
type: Literal["web_search_preview"]
search_context_size: NotRequired[int] # optional
search_context_size: NotRequired[str] # optional
user_location: NotRequired[Union[str, Dict[str, Any]]] # optional - Can be string or structured location
@@ -114,7 +87,7 @@ class CodeInterpreterToolConfig(TypedDict):
"""Configuration for code_interpreter tool."""
type: Literal["code_interpreter"]
container: str # required - Container configuration for code execution
container: str | Dict[str, Any] # required - Container configuration for code execution
class ImageGenerationToolConfig(TypedDict):
@@ -219,24 +192,21 @@ class OpenAIAgentState(BaseModel):
history: List[Dict[str, Any]] = Field(default_factory=list)
# Union type for tool configurations in the config schema
ToolConfigUnion = Union[ComponentModel, BuiltinToolConfig, str]
class OpenAIAgentConfig(BaseModel):
"""Configuration model for OpenAI agent that supports both custom tools and built-in tools.
"""
Configuration model for OpenAI agent supporting OpenAI built-in tools only.
.. versionchanged:: v0.7.0
Added support for built-in tools in JSON configuration via _to_config and _from_config methods.
The tools field now accepts ComponentModel (for custom tools), built-in tool configurations
(dict format), and built-in tool names (string format).
Added support for built-in tools in JSON configuration via _to_config and _from_config methods.
The tools field accepts built-in tool configurations (dict format) and built-in tool names (string format).
Custom tools are not supported.
"""
name: str
description: str
model: str
instructions: str
tools: List[ToolConfigUnion] | None = None
tools: List[Dict[str, Any] | str] | None = None
temperature: Optional[float] = 1
max_output_tokens: Optional[int] = None
json_mode: bool = False
@@ -244,15 +214,6 @@ class OpenAIAgentConfig(BaseModel):
truncation: str = "disabled"
class FunctionExecutionResult(BaseModel):
"""Result of a function execution."""
content: str
call_id: str
name: str
is_error: bool = False
class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
"""
An agent implementation that uses the OpenAI Responses API to generate responses.
@@ -266,26 +227,27 @@ class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
This agent leverages the Responses API to generate responses with capabilities like:
* Custom function calling
* Multi-turn conversations
* Built-in tool support (file_search, code_interpreter, web_search_preview, etc.)
Currently, custom tools are not supported.
.. versionchanged:: v0.7.0
Added support for built-in tool types like file_search, web_search_preview,
code_interpreter, computer_use_preview, image_generation, and mcp.
Added support for tool configurations with required and optional parameters.
Added support for built-in tool types like file_search, web_search_preview,
code_interpreter, computer_use_preview, image_generation, and mcp.
Added support for tool configurations with required and optional parameters.
Built-in tools are split into two categories:
Built-in tools are split into two categories:
**Tools that can use string format** (no required parameters):
**Tools that can use string format** (no required parameters):
- web_search_preview: Can be used as "web_search_preview" or with optional config
(user_location, search_context_size)
- image_generation: Can be used as "image_generation" or with optional config (background, input_image_mask)
- local_shell: Can be used as "local_shell" (WARNING: Only works with codex-mini-latest model)
**Tools that REQUIRE dict configuration** (have required parameters):
**Tools that REQUIRE dict configuration** (have required parameters):
- file_search: MUST use dict with vector_store_ids (List[str])
- computer_use_preview: MUST use dict with display_height (int), display_width (int), environment (str)
@@ -295,6 +257,10 @@ class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
Using required-parameter tools in string format will raise a ValueError with helpful error messages.
The tools parameter type annotation only accepts string values for tools that don't require parameters.
Note:
Custom tools (autogen FunctionTool or other user-defined tools) are not supported by this agent.
Only OpenAI built-in tools provided via the Responses API are supported.
Args:
name (str): Name of the agent
@@ -302,7 +268,7 @@ class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
client (Union[AsyncOpenAI, AsyncAzureOpenAI]): OpenAI client instance
model (str): Model to use (e.g. "gpt-4.1")
instructions (str): System instructions for the agent
tools (Optional[Iterable[Union[str, BuiltinToolConfig, Tool]]]): Tools the agent can use.
tools (Optional[Iterable[Union[str, BuiltinToolConfig]]]): Tools the agent can use.
Supported string values (no required parameters): "web_search_preview", "image_generation", "local_shell".
Dict values can provide configuration for built-in tools with parameters.
Required parameters for built-in tools:
@@ -317,7 +283,7 @@ class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
- mcp: allowed_tools (List[str]), headers (dict), require_approval (bool)
Special tools with model restrictions:
- local_shell: Only works with "codex-mini-latest" model (WARNING: Very limited support)
Also accepts custom Tool objects for function calling.
Custom tools are not supported.
temperature (Optional[float]): Temperature for response generation (default: 1)
max_output_tokens (Optional[int]): Maximum output tokens
json_mode (bool): Whether to use JSON mode (default: False)
@@ -330,66 +296,63 @@ class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
.. code-block:: python
from openai import AsyncOpenAI
from autogen_core import CancellationToken
import asyncio
from autogen_agentchat.ui import Console
from autogen_ext.agents.openai import OpenAIAgent
from autogen_agentchat.messages import TextMessage
import logging
from openai import AsyncOpenAI
async def example():
cancellation_token = CancellationToken()
client = AsyncOpenAI()
agent = OpenAIAgent(
name="Simple Agent",
name="SimpleAgent",
description="A simple OpenAI agent using the Responses API",
client=client,
model="gpt-4o",
model="gpt-4.1",
instructions="You are a helpful assistant.",
tools=["web_search_preview", "image_generation"], # Only tools without required params
tools=["web_search_preview"], # Only tools without required params
)
response = await agent.on_messages(
[TextMessage(source="user", content="Search for recent AI developments")], cancellation_token
)
logging.info(response)
await Console(agent.run_stream(task="Search for recent AI developments"))
asyncio.run(example())
Usage with configured built-in tools:
.. code-block:: python
from openai import AsyncOpenAI
from autogen_core import CancellationToken
import asyncio
from autogen_agentchat.ui import Console
from autogen_ext.agents.openai import OpenAIAgent
from autogen_agentchat.messages import TextMessage
import logging
from openai import AsyncOpenAI
async def example_with_configs():
cancellation_token = CancellationToken()
client = AsyncOpenAI()
# Configure tools with required and optional parameters
tools = [
{
"type": "file_search",
"vector_store_ids": ["vs_abc123"], # required
"max_num_results": 10, # optional
},
{
"type": "computer_use_preview",
"display_height": 1024, # required
"display_width": 1280, # required
"environment": "desktop", # required
},
# {
# "type": "file_search",
# "vector_store_ids": ["vs_abc123"], # required
# "max_num_results": 10, # optional
# },
# {
# "type": "computer_use_preview",
# "display_height": 1024, # required
# "display_width": 1280, # required
# "environment": "linux", # required
# },
{
"type": "code_interpreter",
"container": "python-3.11", # required
},
{
"type": "mcp",
"server_label": "my-mcp-server", # required
"server_url": "http://localhost:3000", # required
"container": {"type": "auto"}, # required
},
# {
# "type": "mcp",
# "server_label": "my-mcp-server", # required
# "server_url": "http://localhost:3000", # required
# },
{
"type": "web_search_preview",
"user_location": { # optional - structured location
@@ -398,73 +361,27 @@ class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
"region": "CA", # optional
"city": "San Francisco", # optional
},
"search_context_size": 5, # optional
"search_context_size": "low", # optional
},
"image_generation", # Simple tools can still use string format
# "image_generation", # Simple tools can still use string format
]
agent = OpenAIAgent(
name="Configured Agent",
name="ConfiguredAgent",
description="An agent with configured tools",
client=client,
model="gpt-4o",
model="gpt-4.1",
instructions="You are a helpful assistant with specialized tools.",
tools=tools, # type: ignore
)
response = await agent.on_messages(
[TextMessage(source="user", content="Search for recent AI developments")], cancellation_token
)
logging.info(response)
Mixed usage with custom function tools:
.. code-block:: python
import asyncio
import logging
from openai import AsyncOpenAI
from autogen_core import CancellationToken
from autogen_ext.agents.openai import OpenAIAgent
from autogen_agentchat.messages import TextMessage
from autogen_core.tools import FunctionTool
await Console(agent.run_stream(task="Search for recent AI developments"))
# Define a simple calculator function
async def calculate(a: int, b: int) -> int:
'''Simple function to add two numbers.'''
return a + b
asyncio.run(example_with_configs())
# Wrap the calculate function as a tool
calculator = FunctionTool(calculate, description="A simple calculator tool")
async def example_mixed_tools():
cancellation_token = CancellationToken()
client = AsyncOpenAI()
# Use the FunctionTool instance defined above
agent = OpenAIAgent(
name="Mixed Tools Agent",
description="An agent with both built-in and custom tools",
client=client,
model="gpt-4o",
instructions="You are a helpful assistant with calculation and web search capabilities.",
tools=[
"web_search_preview",
calculator,
{"type": "mcp", "server_label": "tools", "server_url": "http://localhost:3000"},
],
)
response = await agent.on_messages(
[TextMessage(source="user", content="What's 2+2 and what's the weather like?")],
cancellation_token,
)
logging.info(response)
asyncio.run(example_mixed_tools())
Note:
Custom tools are not supported by OpenAIAgent. Use only built-in tools from the Responses API.
"""
@@ -483,7 +400,6 @@ class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
Union[
Literal["web_search_preview", "image_generation", "local_shell"],
BuiltinToolConfig,
Tool,
]
]
] = None,
@@ -505,7 +421,6 @@ class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
self._last_response_id: Optional[str] = None
self._message_history: List[Dict[str, Any]] = []
self._tools: List[Dict[str, Any]] = []
self._tool_map: Dict[str, Tool] = {}
if tools is not None:
for tool in tools:
if isinstance(tool, str):
@@ -513,11 +428,7 @@ class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
self._add_builtin_tool(tool)
elif isinstance(tool, dict) and "type" in tool:
# Handle configured built-in tools
self._add_configured_tool(tool)
elif isinstance(tool, Tool):
# Handle custom function tools
self._tools.append(cast(dict[str, Any], _convert_tool_to_function_tool_param(tool)))
self._tool_map[tool.name] = tool
self._tools.append(cast(dict[str, Any], tool))
else:
raise ValueError(f"Unsupported tool type: {type(tool)}")
@@ -556,194 +467,12 @@ class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
"""Get help text for required parameters of a tool."""
help_text = {
"file_search": "vector_store_ids (List[str])",
"code_interpreter": "container (str)",
"code_interpreter": "container (str | dict)",
"computer_use_preview": "display_height (int), display_width (int), environment (str)",
"mcp": "server_label (str), server_url (str)",
}
return help_text.get(tool_name, "unknown parameters")
def _add_configured_tool(self, tool_config: BuiltinToolConfig) -> None:
"""Add a configured built-in tool with parameters."""
tool_type = tool_config.get("type")
if not tool_type:
raise ValueError("Tool configuration must include 'type' field")
# If an identical configuration is already present we simply ignore the new one (keeps API payload minimal)
if cast(Dict[str, Any], tool_config) in self._tools:
return
# Initialize tool definition
tool_def: Dict[str, Any] = {}
# Special validation for model-restricted tools
if tool_type == "local_shell":
if self._model != "codex-mini-latest":
raise ValueError(
f"Tool 'local_shell' is only supported with model 'codex-mini-latest', "
f"but current model is '{self._model}'. "
f"This tool is available exclusively through the Responses API and has severe limitations. "
f"Consider using autogen_ext.tools.code_execution.PythonCodeExecutionTool with "
f"autogen_ext.code_executors.local.LocalCommandLineCodeExecutor for shell execution instead."
)
tool_def = {"type": "local_shell"}
# For Responses API, built-in tools are defined directly without nesting
elif tool_type == "file_search":
# file_search requires vector_store_ids
fs_config = cast(FileSearchToolConfig, tool_config)
if "vector_store_ids" not in fs_config:
raise ValueError("file_search tool requires 'vector_store_ids' parameter")
vector_store_ids = fs_config["vector_store_ids"]
if not isinstance(vector_store_ids, list) or not vector_store_ids:
raise ValueError("file_search 'vector_store_ids' must be a non-empty list of strings")
if not all(isinstance(vid, str) and vid.strip() for vid in vector_store_ids):
raise ValueError("file_search 'vector_store_ids' must contain non-empty strings")
tool_def = {"type": "file_search", "vector_store_ids": vector_store_ids}
# Optional parameters
if "max_num_results" in fs_config:
max_results = fs_config["max_num_results"]
if not isinstance(max_results, int) or max_results <= 0:
raise ValueError("file_search 'max_num_results' must be a positive integer")
tool_def["max_num_results"] = max_results
if "ranking_options" in fs_config:
tool_def["ranking_options"] = fs_config["ranking_options"]
if "filters" in fs_config:
tool_def["filters"] = fs_config["filters"]
elif tool_type == "web_search_preview":
# web_search_preview can have optional parameters
ws_config = cast(WebSearchToolConfig, tool_config)
tool_def = {"type": "web_search_preview"}
if "search_context_size" in ws_config:
context_size = ws_config["search_context_size"]
if not isinstance(context_size, int) or context_size <= 0:
raise ValueError("web_search_preview 'search_context_size' must be a positive integer")
tool_def["search_context_size"] = context_size
if "user_location" in ws_config:
user_location = ws_config["user_location"]
if isinstance(user_location, str):
if not user_location.strip():
raise ValueError(
"web_search_preview 'user_location' must be a non-empty string when using string format"
)
elif isinstance(user_location, dict):
if "type" not in user_location:
raise ValueError("web_search_preview 'user_location' dictionary must include 'type' field")
location_type = user_location["type"]
if location_type not in ["approximate", "exact"]:
raise ValueError("web_search_preview 'user_location' type must be 'approximate' or 'exact'")
# Optional fields: country, region, city can be validated if present
for optional_field in ["country", "region", "city"]:
if optional_field in user_location:
if (
not isinstance(user_location[optional_field], str)
or not user_location[optional_field].strip()
):
raise ValueError(
f"web_search_preview 'user_location' {optional_field} must be a non-empty string"
)
else:
raise ValueError("web_search_preview 'user_location' must be a string or dictionary")
tool_def["user_location"] = user_location
elif tool_type == "computer_use_preview":
# computer_use_preview requires display dimensions and environment
cu_config = cast(ComputerUseToolConfig, tool_config)
required_params = ["display_height", "display_width", "environment"]
for param in required_params:
if param not in cu_config:
raise ValueError(f"computer_use_preview tool requires '{param}' parameter")
# Validate display dimensions
height = cu_config["display_height"]
width = cu_config["display_width"]
if not isinstance(height, int) or height <= 0:
raise ValueError("computer_use_preview 'display_height' must be a positive integer")
if not isinstance(width, int) or width <= 0:
raise ValueError("computer_use_preview 'display_width' must be a positive integer")
# Validate environment
environment = cu_config["environment"]
if not isinstance(environment, str) or not environment.strip():
raise ValueError("computer_use_preview 'environment' must be a non-empty string")
tool_def = {
"type": "computer_use_preview",
"display_height": height,
"display_width": width,
"environment": environment,
}
elif tool_type == "mcp":
# MCP requires server_label and server_url
mcp_config = cast(MCPToolConfig, tool_config)
required_params = ["server_label", "server_url"]
for param in required_params:
if param not in mcp_config:
raise ValueError(f"mcp tool requires '{param}' parameter")
# Validate required parameters
server_label = mcp_config["server_label"]
server_url = mcp_config["server_url"]
if not isinstance(server_label, str) or not server_label.strip():
raise ValueError("mcp 'server_label' must be a non-empty string")
if not isinstance(server_url, str) or not server_url.strip():
raise ValueError("mcp 'server_url' must be a non-empty string")
tool_def = {"type": "mcp", "server_label": server_label, "server_url": server_url}
# Optional parameters
if "allowed_tools" in mcp_config:
allowed_tools = mcp_config["allowed_tools"]
if not isinstance(allowed_tools, list):
raise ValueError("mcp 'allowed_tools' must be a list of strings")
if not all(isinstance(tool, str) for tool in allowed_tools):
raise ValueError("mcp 'allowed_tools' must contain only strings")
tool_def["allowed_tools"] = allowed_tools
if "headers" in mcp_config:
headers = mcp_config["headers"]
if not isinstance(headers, dict):
raise ValueError("mcp 'headers' must be a dictionary")
tool_def["headers"] = headers
if "require_approval" in mcp_config:
require_approval = mcp_config["require_approval"]
if not isinstance(require_approval, bool):
raise ValueError("mcp 'require_approval' must be a boolean")
tool_def["require_approval"] = require_approval
elif tool_type == "code_interpreter":
# code_interpreter requires container
ci_config = cast(CodeInterpreterToolConfig, tool_config)
if "container" not in ci_config:
raise ValueError("code_interpreter tool requires 'container' parameter")
container = ci_config["container"]
if not isinstance(container, str) or not container.strip():
raise ValueError("code_interpreter 'container' must be a non-empty string")
tool_def = {"type": "code_interpreter", "container": container}
elif tool_type == "image_generation":
# image_generation can have optional parameters
ig_config = cast(ImageGenerationToolConfig, tool_config)
tool_def = {"type": "image_generation"}
if "background" in ig_config:
background = ig_config["background"]
if not isinstance(background, str) or not background.strip():
raise ValueError("image_generation 'background' must be a non-empty string")
tool_def["background"] = background
if "input_image_mask" in ig_config:
input_image_mask = ig_config["input_image_mask"]
if not isinstance(input_image_mask, str) or not input_image_mask.strip():
raise ValueError("image_generation 'input_image_mask' must be a non-empty string")
tool_def["input_image_mask"] = input_image_mask
else:
raise ValueError(f"Unsupported built-in tool type: {tool_type}")
self._tools.append(tool_def)
def _convert_message_to_dict(self, message: OpenAIMessage) -> Dict[str, Any]:
"""Convert an OpenAIMessage to a Dict[str, Any]."""
return dict(message)
@@ -763,38 +492,7 @@ class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
"""Return the types of messages that this agent can produce."""
return [TextMessage, MultiModalMessage, StopMessage, ToolCallSummaryMessage, HandoffMessage]
async def _execute_tool_call(
self: "OpenAIAgent", tool_call: FunctionCall, cancellation_token: CancellationToken
) -> FunctionExecutionResult:
tool_name = tool_call.name
if tool_name not in self._tool_map:
return FunctionExecutionResult(
content=f"Error: Tool '{tool_name}' is not available",
call_id=tool_call.id,
name=tool_name,
is_error=True,
)
tool = self._tool_map[tool_name]
try:
try:
arguments = json.loads(tool_call.arguments)
except json.JSONDecodeError as json_err:
return FunctionExecutionResult(
content=f"Error: Invalid JSON in tool arguments - {str(json_err)}",
call_id=tool_call.id,
name=tool_name,
is_error=True,
)
result = await tool.run_json(arguments, cancellation_token, call_id=tool_call.id)
return FunctionExecutionResult(
content=tool.return_value_as_string(result), call_id=tool_call.id, name=tool_name, is_error=False
)
except Exception as e:
error_msg = f"Error: {str(e)}"
event_logger.warning(f"Tool execution error in {tool_name}: {error_msg}")
return FunctionExecutionResult(content=error_msg, call_id=tool_call.id, name=tool_name, is_error=True)
# Custom tool execution is not supported by this agent.
def _build_api_parameters(self: "OpenAIAgent", messages: List[Dict[str, Any]]) -> Dict[str, Any]:
has_system_message = any(msg.get("role") == "system" for msg in messages)
@@ -915,73 +613,17 @@ class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
def _to_config(self: "OpenAIAgent") -> OpenAIAgentConfig:
"""Convert the OpenAI agent to a declarative config.
Serializes both custom Tool objects and built-in tools to their appropriate
configuration formats for JSON serialization.
.. versionchanged:: v0.6.2
Added support for serializing built-in tools alongside custom tools.
Serializes built-in tools to their appropriate configuration formats for JSON serialization.
Returns:
OpenAIAgentConfig: The configuration that can recreate this agent.
"""
# Serialize tools in the **original order** they were registered. We iterate over the
# internal ``self._tools`` list which contains both built-in tool definitions **and** the
# synthetic "function" records for custom :class:`Tool` objects. For the latter we
# convert the synthetic record back to a :class:`ComponentModel` by looking up the actual
# tool instance in ``self._tool_map``. This approach keeps ordering stable while still
# supporting full round-trip serialisation.
tool_configs: List[ToolConfigUnion] = []
for tool_def in self._tools:
# 1. Custom function tools are stored internally as ``{"type": "function", "function": {...}}``.
if tool_def.get("type") == "function":
fn_schema = cast(Dict[str, Any], tool_def.get("function", {}))
tool_name = fn_schema.get("name") # type: ignore[arg-type]
if tool_name and tool_name in self._tool_map:
tool_obj = self._tool_map[tool_name]
try:
if hasattr(tool_obj, "dump_component"):
component_model = cast(Any, tool_obj).dump_component()
tool_configs.append(component_model)
else:
component_model = ComponentModel(
provider="autogen_core.tools.FunctionTool",
component_type=None,
config={
"name": tool_obj.name,
"description": getattr(tool_obj, "description", ""),
},
)
tool_configs.append(component_model)
except Exception as e: # pragma: no cover – extremely unlikely
warnings.warn(
f"Error serializing tool '{tool_name}': {e}",
stacklevel=2,
)
component_model = ComponentModel(
provider="autogen_core.tools.FunctionTool",
component_type=None,
config={
"name": tool_name or "unknown_tool",
"description": getattr(tool_obj, "description", ""),
},
)
tool_configs.append(component_model)
# 2. Built-in tools are already in their correct dict form – append verbatim.
elif "type" in tool_def: # built-in tool
tool_configs.append(cast(BuiltinToolConfig, tool_def))
else: # pragma: no cover – should never happen
warnings.warn(
f"Encountered unexpected tool definition during serialisation: {tool_def}",
stacklevel=2,
)
return OpenAIAgentConfig(
name=self.name,
description=self.description,
model=self._model,
instructions=self._instructions,
tools=tool_configs if tool_configs else None,
tools=list(self._tools),
temperature=self._temperature,
max_output_tokens=self._max_output_tokens,
json_mode=self._json_mode,
@@ -993,80 +635,25 @@ class OpenAIAgent(BaseChatAgent, Component[OpenAIAgentConfig]):
def _from_config(cls: Type["OpenAIAgent"], config: OpenAIAgentConfig) -> "OpenAIAgent":
"""Create an OpenAI agent from a declarative config.
Handles both custom Tool objects (from ComponentModel) and built-in tools
(from string or dict configurations).
Handles built-in tools (from string or dict configurations).
.. versionchanged:: v0.6.2
Added support for loading built-in tools alongside custom tools.
Args:
config: The configuration to load the agent from.
Args:
config: The configuration to load the agent from.
Returns:
OpenAIAgent: The reconstructed agent.
Returns:
OpenAIAgent: The reconstructed agent.
"""
from openai import AsyncOpenAI
client = AsyncOpenAI()
tools: Optional[List[Union[str, BuiltinToolConfig, Tool]]] = None
if config.tools:
tools_list: List[Union[str, BuiltinToolConfig, Tool]] = []
for tool_config in config.tools:
# Handle ComponentModel (custom Tool objects)
if isinstance(tool_config, ComponentModel):
try:
provider = tool_config.provider
module_name, class_name = provider.rsplit(".", 1)
module = __import__(module_name, fromlist=[class_name])
tool_cls = getattr(module, class_name)
tool = tool_cls(**tool_config.config)
tools_list.append(cast(Tool, tool))
except Exception as e:
warnings.warn(f"Error loading custom tool: {e}", stacklevel=2)
from autogen_core.tools import FunctionTool
async def dummy_func(*args: Any, **kwargs: Any) -> str:
return "Tool not fully restored"
tool = FunctionTool(
name=tool_config.config.get("name", "unknown_tool"),
description=tool_config.config.get("description", ""),
func=dummy_func,
)
tools_list.append(tool)
# Handle string format built-in tools
elif isinstance(tool_config, str):
tools_list.append(tool_config)
# Handle dict format built-in tools
elif isinstance(tool_config, dict) and "type" in tool_config:
tools_list.append(tool_config) # type: ignore[arg-type]
else:
warnings.warn(f"Unknown tool configuration format: {type(tool_config)}", stacklevel=2)
tools = tools_list if tools_list else None
return cls(
name=config.name,
description=config.description,
client=client,
model=config.model,
instructions=config.instructions,
tools=cast(
Optional[
Iterable[
Union[
BuiltinToolConfig,
Tool,
Literal["web_search_preview", "image_generation", "local_shell"],
]
]
],
tools,
),
tools=config.tools, # type: ignore
temperature=config.temperature,
max_output_tokens=config.max_output_tokens,
json_mode=config.json_mode,

80
python/packages/autogen-ext/tests/agents/test_openai_agent_builtin_tool_validation.py
View File

@@ -85,80 +85,6 @@ async def test_builtin_tool_string_validation(
assert any(t["type"] == tool_name for t in agent.tools)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"tool_config,should_raise",
[
# file_search: missing required param
({"type": "file_search"}, True),
# file_search: empty vector_store_ids
({"type": "file_search", "vector_store_ids": []}, True),
# file_search: invalid type
({"type": "file_search", "vector_store_ids": [123]}, True),
# file_search: valid
({"type": "file_search", "vector_store_ids": ["vs1"]}, False),
# computer_use_preview: missing param
({"type": "computer_use_preview", "display_height": 100, "display_width": 100}, True),
# computer_use_preview: invalid type
({"type": "computer_use_preview", "display_height": -1, "display_width": 100, "environment": "desktop"}, True),
# computer_use_preview: valid
(
{"type": "computer_use_preview", "display_height": 100, "display_width": 100, "environment": "desktop"},
False,
),
# code_interpreter: missing param
({"type": "code_interpreter"}, True),
# code_interpreter: empty container
({"type": "code_interpreter", "container": ""}, True),
# code_interpreter: valid
({"type": "code_interpreter", "container": "python-3.11"}, False),
# mcp: missing param
({"type": "mcp", "server_label": "label"}, True),
# mcp: invalid type
({"type": "mcp", "server_label": "", "server_url": "url"}, True),
# mcp: valid
({"type": "mcp", "server_label": "label", "server_url": "url"}, False),
# web_search_preview: valid with string user_location
({"type": "web_search_preview", "user_location": "US"}, False),
# web_search_preview: valid with dict user_location
({"type": "web_search_preview", "user_location": {"type": "approximate"}}, False),
# web_search_preview: invalid user_location type
({"type": "web_search_preview", "user_location": 123}, True),
# image_generation: valid with background
({"type": "image_generation", "background": "white"}, False),
# image_generation: invalid background
({"type": "image_generation", "background": ""}, True),
],
)
async def test_builtin_tool_dict_validation(
tool_config: Dict[str, Any], should_raise: bool, openai_client: AsyncOpenAI
) -> None:
"""Test validation of dictionary-based builtin tools."""
client = openai_client
tools = [tool_config] # type: ignore
if should_raise:
with pytest.raises(ValueError):
OpenAIAgent(
name="test",
description="desc",
client=client,
model="gpt-4o",
instructions="inst",
tools=tools, # type: ignore
)
else:
agent = OpenAIAgent(
name="test",
description="desc",
client=client,
model="gpt-4o",
instructions="inst",
tools=tools, # type: ignore
)
assert any(t["type"] == tool_config["type"] for t in agent.tools)
@pytest.mark.asyncio
async def test_builtin_tool_validation_with_custom_and_builtin(openai_client: AsyncOpenAI) -> None:
"""Test validation with mixed string and dictionary tools."""
@@ -478,7 +404,7 @@ async def test_to_config_with_string_builtin_tools() -> None:
if isinstance(tool, str):
tool_types.append(tool)
elif isinstance(tool, dict):
tool_types.append(cast(Dict[str, Any], tool)["type"])
tool_types.append(tool["type"])
else:
# Handle ComponentModel case
tool_types.append(str(tool))
@@ -510,7 +436,7 @@ async def test_to_config_with_configured_builtin_tools() -> None:
assert len(config.tools) == 3
# Verify configured tools are serialized correctly
tool_configs = [cast(Dict[str, Any], tool) for tool in config.tools if isinstance(tool, dict)]
tool_configs = [tool for tool in config.tools if isinstance(tool, dict)]
assert len(tool_configs) == 3
# Check file_search config
@@ -657,7 +583,7 @@ async def test_config_serialization_with_mixed_tools() -> None:
assert len(config.tools) == 4
# Verify all tools are serialized as dicts with "type" key
dict_tools = [cast(Dict[str, Any], tool) for tool in config.tools if isinstance(tool, dict)]
dict_tools = [tool for tool in config.tools if isinstance(tool, dict)]
assert len(dict_tools) == 4
# Check that string tools are converted to dicts with "type" key

215
python/packages/autogen-ext/tests/test_openai_agent.py
View File

@@ -1,5 +1,4 @@
import json
from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional, Type, Union, cast
from typing import Any, AsyncGenerator, List, Union, cast
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -7,42 +6,8 @@ from autogen_agentchat.base import Response
from autogen_agentchat.messages import BaseChatMessage, MultiModalMessage, TextMessage
from autogen_core import CancellationToken, Image
from autogen_core.models import UserMessage
from autogen_core.tools import Tool, ToolSchema
from autogen_ext.agents.openai import OpenAIAgent
from openai import AsyncOpenAI
from pydantic import BaseModel
class FakeChunkDelta:
def __init__(self, content: Optional[str] = None, tool_calls: Optional[List[Any]] = None) -> None:
self.content = content
self.tool_calls = tool_calls
class FakeChunkChoice:
def __init__(self, delta: Optional[FakeChunkDelta] = None, finish_reason: Optional[str] = None) -> None:
self.delta = delta
self.finish_reason = finish_reason
self.index = 0
class FakeChunk:
def __init__(self, id: str = "chunk-1", choices: Optional[List[FakeChunkChoice]] = None) -> None:
self.id = id
self.choices = choices or []
class FakeToolCallFunction:
def __init__(self, name: str = "", arguments: str = "") -> None:
self.name = name
self.arguments = arguments
class FakeToolCall:
def __init__(self, id: str = "call-1", function: Optional[FakeToolCallFunction] = None) -> None:
self.id = id
self.type = "function"
self.function = function or FakeToolCallFunction()
def create_mock_openai_client() -> AsyncOpenAI:
@@ -92,92 +57,15 @@ def cancellation_token() -> CancellationToken:
return CancellationToken()
class WeatherResponse(BaseModel):
temperature: float
conditions: str
class GetWeatherArgs(BaseModel):
location: str
class WeatherTool(Tool):
def __init__(self) -> None:
self._name = "get_weather"
self._description = "Get the current weather in a location"
self._input_schema = GetWeatherArgs
self._output_schema = WeatherResponse
self._schema = ToolSchema(
name=self._name,
description=self._description,
parameters={
"type": "object",
"properties": {"location": {"type": "string", "description": "The location to get weather for"}},
"required": ["location"],
},
)
@property
def name(self) -> str:
return self._name
@property
def description(self) -> str:
return "Get the current weather in a location"
@property
def schema(self) -> ToolSchema:
return self._schema
def args_type(self) -> Type[BaseModel]:
return GetWeatherArgs
def return_type(self) -> Type[Any]:
return WeatherResponse
def state_type(self) -> Type[BaseModel] | None:
return None
def return_value_as_string(self, value: Any) -> str:
if isinstance(value, dict):
return json.dumps(value)
return str(value)
async def run_json(
self, args: Mapping[str, Any], cancellation_token: CancellationToken, call_id: str | None = None
) -> Dict[str, Any]:
_ = GetWeatherArgs(**args)
return WeatherResponse(temperature=72.5, conditions="sunny").model_dump()
async def load_state_json(self, state: Mapping[str, Any]) -> None:
pass
async def save_state_json(self) -> Mapping[str, Any]:
return {}
@pytest.fixture
def weather_tool() -> WeatherTool:
return WeatherTool()
@pytest.fixture
def failing_tool() -> Tool:
tool = MagicMock(spec=Tool)
tool.name = "failing_tool"
tool.run_json = AsyncMock(side_effect=Exception("Tool execution failed"))
return tool
@pytest.fixture
def agent(mock_openai_client: AsyncOpenAI, weather_tool: WeatherTool) -> OpenAIAgent:
def agent(mock_openai_client: AsyncOpenAI) -> OpenAIAgent:
return OpenAIAgent(
name="assistant",
description="Test assistant using the Response API",
client=mock_openai_client,
model="gpt-4o",
instructions="You are a helpful AI assistant.",
tools=[weather_tool],
tools=["web_search_preview"],
temperature=0.7,
max_output_tokens=1000,
store=True,
@@ -222,27 +110,19 @@ async def test_basic_response(agent: OpenAIAgent, cancellation_token: Cancellati
@pytest.mark.asyncio
async def test_tool_calling(agent: OpenAIAgent, cancellation_token: CancellationToken) -> None:
"""Test that the agent can call a tool and return the result using the Responses API."""
"""Test that enabling a built-in tool yields a tool-style JSON response via the Responses API."""
message = TextMessage(source="user", content="What's the weather in New York?")
async def mock_run_json(self: Any, args: Dict[str, Any], cancellation_token: CancellationToken) -> Dict[str, Any]:
return {"temperature": 75.0, "conditions": "sunny and clear"}
all_messages: List[Any] = []
async for msg in agent.on_messages_stream([message], cancellation_token):
all_messages.append(msg)
with patch.object(WeatherTool, "run_json", mock_run_json):
message = TextMessage(source="user", content="What's the weather in New York?")
all_messages: List[Any] = []
async for msg in agent.on_messages_stream([message], cancellation_token):
all_messages.append(msg)
final_response = next((msg for msg in all_messages if hasattr(msg, "chat_message")), None)
assert final_response is not None
assert hasattr(final_response, "chat_message")
response_msg = cast(Response, final_response)
assert isinstance(response_msg.chat_message, TextMessage)
assert response_msg.chat_message.content in (
'{"temperature": 75.0, "conditions": "sunny and clear"}',
'{"temperature": 72.5, "conditions": "sunny"}',
)
final_response = next((msg for msg in all_messages if hasattr(msg, "chat_message")), None)
assert final_response is not None
assert hasattr(final_response, "chat_message")
response_msg = cast(Response, final_response)
assert isinstance(response_msg.chat_message, TextMessage)
assert response_msg.chat_message.content == '{"temperature": 72.5, "conditions": "sunny"}'
@pytest.mark.asyncio
@@ -312,19 +192,6 @@ async def test_convert_message_functions(agent: OpenAIAgent) -> None:
assert openai_text_msg["content"] == "Plain text"
@pytest.mark.asyncio
async def test_tool_schema_conversion(agent: OpenAIAgent) -> None:
from autogen_ext.agents.openai._openai_agent import _convert_tool_to_function_tool_param # type: ignore
tool_schema = _convert_tool_to_function_tool_param(agent._tool_map["get_weather"]) # type: ignore
assert tool_schema["name"] == "get_weather"
assert "description" in tool_schema
assert "parameters" in tool_schema and isinstance(tool_schema["parameters"], dict)
assert tool_schema["parameters"].get("type") == "object"
assert "properties" in tool_schema["parameters"]
@pytest.mark.asyncio
async def test_on_messages_inner_messages(agent: OpenAIAgent, cancellation_token: CancellationToken) -> None:
class DummyMsg(BaseChatMessage):
@@ -408,20 +275,7 @@ async def test_on_messages_stream(agent: OpenAIAgent, cancellation_token: Cancel
@pytest.mark.asyncio
async def test_component_serialization(agent: OpenAIAgent) -> None:
config = agent.dump_component()
config_dict: Any = None
if isinstance(config, dict):
config_dict = config
elif hasattr(config, "model_dump_json"):
config_dict = json.loads(config.model_dump_json())
elif hasattr(config, "model_dump"):
config_dict = config.model_dump()
elif isinstance(config, str):
config_dict = json.loads(config)
else:
config_dict = {"name": agent.name, "description": agent.description}
if isinstance(config_dict, dict) and "config" in config_dict:
config_dict = config_dict["config"]
config_dict = config.config
assert config_dict["name"] == "assistant"
assert config_dict["description"] == "Test assistant using the Response API"
@@ -437,45 +291,8 @@ async def test_component_serialization(agent: OpenAIAgent) -> None:
async def test_from_config(agent: OpenAIAgent) -> None:
config = agent.dump_component()
config_dict: Dict[str, Any] = {}
if hasattr(config, "model_dump_json"):
config_dict = json.loads(config.model_dump_json())
elif isinstance(config, str):
config_dict = json.loads(config)
elif isinstance(config, dict):
config_dict = config
if "tools" in config_dict and config_dict["tools"] is not None:
serialized_tools: List[Dict[str, Any]] = []
tools_any: Any = config_dict["tools"]
if isinstance(tools_any, list):
tools_list: List[Any] = cast(List[Any], tools_any) # type: ignore[redundant-cast]
tools_count: int = len(tools_list)
for i in range(tools_count):
tool_any: Any = tools_list[i]
tool_dict: Dict[str, Any] = {}
if isinstance(tool_any, dict):
tool_dict = tool_any
elif tool_any is not None and isinstance(tool_any, object) and hasattr(tool_any, "model_dump"):
model_dump_any: Any = getattr(tool_any, "model_dump", None)
if callable(model_dump_any):
try:
result_any: Any = model_dump_any()
if isinstance(result_any, dict):
tool_dict = result_any
else:
tool_dict = {"provider": "unknown", "config": {}}
except Exception:
tool_dict = {"provider": "unknown", "config": {}}
else:
tool_dict = {"provider": "unknown", "config": {}}
else:
tool_dict = {"provider": "unknown", "config": {}}
serialized_tools.append(tool_dict)
config_dict["tools"] = serialized_tools
with patch("openai.AsyncOpenAI"):
loaded_agent = OpenAIAgent.load_component(config_dict)
loaded_agent = OpenAIAgent.load_component(config)
assert loaded_agent.name == "assistant"
assert loaded_agent.description == "Test assistant using the Response API"