autogen_ext.agents.openai#

class OpenAIAgent(name: str, description: str, client: AsyncOpenAI | AsyncAzureOpenAI, model: str, instructions: str, tools: Iterable[Literal['web_search_preview', 'image_generation', 'local_shell'] | FileSearchToolConfig | WebSearchToolConfig | ComputerUseToolConfig | MCPToolConfig | CodeInterpreterToolConfig | ImageGenerationToolConfig | LocalShellToolConfig] | None = None, temperature: float | None = 1, max_output_tokens: int | None = None, json_mode: bool = False, store: bool = True, truncation: str = 'disabled')[source]#

Bases: BaseChatAgent, Component[OpenAIAgentConfig]

An agent implementation that uses the OpenAI Responses API to generate responses.

Installation:

pip install "autogen-ext[openai]"
# pip install "autogen-ext[openai,azure]"  # For Azure OpenAI Assistant

This agent leverages the Responses API to generate responses with capabilities like:

  • Multi-turn conversations

  • Built-in tool support (file_search, code_interpreter, web_search_preview, etc.)

Currently, custom tools are not supported.

Changed in version 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.

Built-in tools are split into two categories:

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):

  • 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)

  • code_interpreter: MUST use dict with container (str)

  • mcp: MUST use dict with server_label (str), server_url (str)

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.

Parameters:

  • name (str) – Name of the agent

  • description (str) – Description of the agent’s purpose

  • 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]]]) – 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: - file_search: vector_store_ids (List[str]) - computer_use_preview: display_height (int), display_width (int), environment (str) - code_interpreter: container (str) - mcp: server_label (str), server_url (str) Optional parameters for built-in tools: - file_search: max_num_results (int), ranking_options (dict), filters (dict) - web_search_preview: user_location (str or dict), search_context_size (int) - image_generation: background (str), input_image_mask (str) - 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) 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)

  • store (bool) – Whether to store conversations (default: True)

  • truncation (str) – Truncation strategy (default: “disabled”)

Example

Basic usage with built-in tools:

import asyncio

from autogen_agentchat.ui import Console
from autogen_ext.agents.openai import OpenAIAgent
from openai import AsyncOpenAI


async def example():
    client = AsyncOpenAI()
    agent = OpenAIAgent(
        name="SimpleAgent",
        description="A simple OpenAI agent using the Responses API",
        client=client,
        model="gpt-4.1",
        instructions="You are a helpful assistant.",
        tools=["web_search_preview"],  # Only tools without required params
    )
    await Console(agent.run_stream(task="Search for recent AI developments"))


asyncio.run(example())

Usage with configured built-in tools:

import asyncio

from autogen_agentchat.ui import Console
from autogen_ext.agents.openai import OpenAIAgent
from openai import AsyncOpenAI


async def example_with_configs():
    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": "linux",  # required
        # },
        {
            "type": "code_interpreter",
            "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
                "type": "approximate",  # required: "approximate" or "exact"
                "country": "US",  # optional
                "region": "CA",  # optional
                "city": "San Francisco",  # optional
            },
            "search_context_size": "low",  # optional
        },
        # "image_generation",  # Simple tools can still use string format
    ]

    agent = OpenAIAgent(
        name="ConfiguredAgent",
        description="An agent with configured tools",
        client=client,
        model="gpt-4.1",
        instructions="You are a helpful assistant with specialized tools.",
        tools=tools,  # type: ignore
    )
    await Console(agent.run_stream(task="Search for recent AI developments"))


asyncio.run(example_with_configs())
Note:

Custom tools are not supported by OpenAIAgent. Use only built-in tools from the Responses API.

component_config_schema#

alias of OpenAIAgentConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.agents.openai.OpenAIAgent'#

Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

property produced_message_types: Sequence[Type[TextMessage] | Type[MultiModalMessage] | Type[StopMessage] | Type[ToolCallSummaryMessage] | Type[HandoffMessage]]#

Return the types of messages that this agent can produce.

async on_messages(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) → Response[source]#

Handles incoming messages and returns a response.

Note

Agents are stateful and the messages passed to this method should be the new messages since the last call to this method. The agent should maintain its state between calls to this method. For example, if the agent needs to remember the previous messages to respond to the current message, it should store the previous messages in the agent state.

async on_messages_stream(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) → AsyncGenerator[Annotated[ToolCallRequestEvent | ToolCallExecutionEvent | MemoryQueryEvent | UserInputRequestedEvent | ModelClientStreamingChunkEvent | ThoughtEvent | SelectSpeakerEvent | CodeGenerationEvent | CodeExecutionEvent, FieldInfo(annotation=NoneType, required=True, discriminator='type')] | TextMessage | MultiModalMessage | StopMessage | ToolCallSummaryMessage | HandoffMessage | Response, None][source]#

Handles incoming messages and returns a stream of messages and and the final item is the response. The base implementation in BaseChatAgent simply calls on_messages() and yields the messages in the response.

Note

Agents are stateful and the messages passed to this method should be the new messages since the last call to this method. The agent should maintain its state between calls to this method. For example, if the agent needs to remember the previous messages to respond to the current message, it should store the previous messages in the agent state.

async on_reset(cancellation_token: CancellationToken) → None[source]#

Resets the agent to its initialization state.

async save_state() → Mapping[str, Any][source]#

Export state. Default implementation for stateless agents.

async load_state(state: Mapping[str, Any]) → None[source]#

Restore agent from saved state. Default implementation for stateless agents.

to_config() → OpenAIAgentConfig[source]#

Public wrapper for the private _to_config method.

classmethod from_config(config: OpenAIAgentConfig) → OpenAIAgent[source]#

Public wrapper for the private _from_config classmethod.

property tools: list[Any]#

Public access to the agent’s tools.

property model: str#

Public access to the agent’s model.

class OpenAIAssistantAgent(name: str, description: str, client: AsyncOpenAI | AsyncAzureOpenAI, model: str, instructions: str, tools: Iterable[Literal['code_interpreter', 'file_search'] | Tool | Callable[[...], Any] | Callable[[...], Awaitable[Any]]] | None = None, assistant_id: str | None = None, thread_id: str | None = None, metadata: Dict[str, str] | None = None, response_format: Literal['auto'] | ResponseFormatText | ResponseFormatJSONObject | ResponseFormatJSONSchema | None = None, temperature: float | None = None, tool_resources: ToolResources | None = None, top_p: float | None = None)[source]#

Bases: BaseChatAgent

An agent implementation that uses the Assistant API to generate responses.

Installation:

pip install "autogen-ext[openai]"  # For OpenAI Assistant
# pip install "autogen-ext[openai,azure]"  # For Azure OpenAI Assistant

This agent leverages the Assistant API to create AI assistants with capabilities like:

  • Code interpretation and execution

  • File handling and search

  • Custom function calling

  • Multi-turn conversations

The agent maintains a thread of conversation and can use various tools including

  • Code interpreter: For executing code and working with files

  • File search: For searching through uploaded documents

  • Custom functions: For extending capabilities with user-defined tools

Key Features:

  • Supports multiple file formats including code, documents, images

  • Can handle up to 128 tools per assistant

  • Maintains conversation context in threads

  • Supports file uploads for code interpreter and search

  • Vector store integration for efficient file search

  • Automatic file parsing and embedding

You can use an existing thread or assistant by providing the thread_id or assistant_id parameters.

Examples

Use the assistant to analyze data in a CSV file:

from openai import AsyncOpenAI
from autogen_core import CancellationToken
import asyncio
from autogen_ext.agents.openai import OpenAIAssistantAgent
from autogen_agentchat.messages import TextMessage


async def example():
    cancellation_token = CancellationToken()

    # Create an OpenAI client
    client = AsyncOpenAI(api_key="your-api-key", base_url="your-base-url")

    # Create an assistant with code interpreter
    assistant = OpenAIAssistantAgent(
        name="PythonHelper",
        description="Helps with Python programming",
        client=client,
        model="gpt-4",
        instructions="You are a helpful Python programming assistant.",
        tools=["code_interpreter"],
    )

    # Upload files for the assistant to use
    await assistant.on_upload_for_code_interpreter("data.csv", cancellation_token)

    # Get response from the assistant
    response = await assistant.on_messages(
        [TextMessage(source="user", content="Analyze the data in data.csv")], cancellation_token
    )

    print(response)

    # Clean up resources
    await assistant.delete_uploaded_files(cancellation_token)
    await assistant.delete_assistant(cancellation_token)


asyncio.run(example())

Use Azure OpenAI Assistant with AAD authentication:

from openai import AsyncAzureOpenAI
import asyncio
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from autogen_core import CancellationToken
from autogen_ext.agents.openai import OpenAIAssistantAgent
from autogen_agentchat.messages import TextMessage


async def example():
    cancellation_token = CancellationToken()

    # Create an Azure OpenAI client
    token_provider = get_bearer_token_provider(DefaultAzureCredential())
    client = AsyncAzureOpenAI(
        azure_deployment="YOUR_AZURE_DEPLOYMENT",
        api_version="YOUR_API_VERSION",
        azure_endpoint="YOUR_AZURE_ENDPOINT",
        azure_ad_token_provider=token_provider,
    )

    # Create an assistant with code interpreter
    assistant = OpenAIAssistantAgent(
        name="PythonHelper",
        description="Helps with Python programming",
        client=client,
        model="gpt-4o",
        instructions="You are a helpful Python programming assistant.",
        tools=["code_interpreter"],
    )

    # Get response from the assistant
    response = await assistant.on_messages([TextMessage(source="user", content="Hello.")], cancellation_token)

    print(response)

    # Clean up resources
    await assistant.delete_assistant(cancellation_token)


asyncio.run(example())
Parameters:

  • name (str) – Name of the assistant

  • description (str) – Description of the assistant’s purpose

  • client (AsyncOpenAI | AsyncAzureOpenAI) – OpenAI client or Azure OpenAI client instance

  • model (str) – Model to use (e.g. “gpt-4”)

  • instructions (str) – System instructions for the assistant

  • tools (Optional[Iterable[Union[Literal["code_interpreter", "file_search"], Tool | Callable[..., Any] | Callable[..., Awaitable[Any]]]]]) – Tools the assistant can use

  • assistant_id (Optional[str]) – ID of existing assistant to use

  • thread_id (Optional[str]) – ID of existing thread to use

  • metadata (Optional[Dict[str, str]]) – Additional metadata for the assistant.

  • response_format (Optional[AssistantResponseFormatOptionParam]) – Response format settings

  • temperature (Optional[float]) – Temperature for response generation

  • tool_resources (Optional[ToolResources]) – Additional tool configuration

  • top_p (Optional[float]) – Top p sampling parameter

property produced_message_types: Sequence[type[BaseChatMessage]]#

The types of messages that the assistant agent produces.

property threads: AsyncThreads#
property runs: AsyncRuns#
property messages: AsyncMessages#
async on_messages(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) → Response[source]#

Handle incoming messages and return a response.

async on_messages_stream(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) → AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None][source]#

Handle incoming messages and return a response.

async handle_incoming_message(message: BaseChatMessage, cancellation_token: CancellationToken) → None[source]#

Handle regular text messages by adding them to the thread.

async on_reset(cancellation_token: CancellationToken) → None[source]#

Handle reset command by deleting new messages and runs since initialization.

async on_upload_for_code_interpreter(file_paths: str | Iterable[str], cancellation_token: CancellationToken) → None[source]#

Handle file uploads for the code interpreter.

Handle file uploads for file search.

async delete_uploaded_files(cancellation_token: CancellationToken) → None[source]#

Delete all files that were uploaded by this agent instance.

async delete_assistant(cancellation_token: CancellationToken) → None[source]#

Delete the assistant if it was created by this instance.

async delete_vector_store(cancellation_token: CancellationToken) → None[source]#

Delete the vector store if it was created by this instance.

async save_state() → Mapping[str, Any][source]#

Export state. Default implementation for stateless agents.

async load_state(state: Mapping[str, Any]) → None[source]#

Restore agent from saved state. Default implementation for stateless agents.