autogen_ext.agents.openai#

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:

Custom function calling
Multi-turn conversations
Built-in tool support (file_search, code_interpreter, web_search_preview, etc.)

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.

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, Tool]]]) – 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) Also accepts custom Tool objects for function calling.
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:

from openai import AsyncOpenAI
from autogen_core import CancellationToken
from autogen_ext.agents.openai import OpenAIAgent
from autogen_agentchat.messages import TextMessage
import logging


async def example():
    cancellation_token = CancellationToken()
    client = AsyncOpenAI()
    agent = OpenAIAgent(
        name="Simple Agent",
        description="A simple OpenAI agent using the Responses API",
        client=client,
        model="gpt-4o",
        instructions="You are a helpful assistant.",
        tools=["web_search_preview", "image_generation"],  # Only tools without required params
    )
    response = await agent.on_messages(
        [TextMessage(source="user", content="Search for recent AI developments")], cancellation_token
    )
    logging.info(response)

Usage with configured built-in tools:

from openai import AsyncOpenAI
from autogen_core import CancellationToken
from autogen_ext.agents.openai import OpenAIAgent
from autogen_agentchat.messages import TextMessage
import logging


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": "code_interpreter",
            "container": "python-3.11",  # 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": 5,  # optional
        },
        "image_generation",  # Simple tools can still use string format
    ]

    agent = OpenAIAgent(
        name="Configured Agent",
        description="An agent with configured tools",
        client=client,
        model="gpt-4o",
        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:

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


# Define a simple calculator function
async def calculate(a: int, b: int) -> int:
    '''Simple function to add two numbers.'''
    return a + b


# 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())

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.

async list_assistants(after: str | None = None, before: str | None = None, limit: int | None = 20, order: str | None = 'desc') → Dict[str, Any][source]#

List all assistants using the OpenAI API.

Parameters:

after (Optional[str]) – Cursor for pagination (fetch after this assistant ID).
before (Optional[str]) – Cursor for pagination (fetch before this assistant ID).
limit (Optional[int]) – Number of assistants to return (1-100, default 20).
order (Optional[str]) – ‘asc’ or ‘desc’ by created_at (default ‘desc’).

Returns:

Dict[str, Any] – The OpenAI API response containing: - object: ‘list’ - data: List of assistant objects - first_id: str - last_id: str - has_more: bool

Example

import asyncio
from typing import Dict, Any
from autogen_ext.agents.openai import OpenAIAgent
from openai import AsyncOpenAI
import logging


async def example() -> None:
    client = AsyncOpenAI()
    agent = OpenAIAgent(
        name="test_agent",
        description="Test agent",
        client=client,
        model="gpt-4",
        instructions="You are a helpful assistant.",
    )
    assistants: Dict[str, Any] = await agent.list_assistants(limit=5)
    logging.info(assistants)


asyncio.run(example())

async retrieve_assistant(assistant_id: str) → Dict[str, Any][source]#

Retrieve a single assistant by its ID using the OpenAI API.

Parameters:: assistant_id (str) – The ID of the assistant to retrieve.
Returns:: Dict[str, Any] – The assistant object.

Example

import asyncio
from typing import Dict, Any
from autogen_ext.agents.openai import OpenAIAgent
from openai import AsyncOpenAI
import logging


async def example() -> None:
    client = AsyncOpenAI()
    agent = OpenAIAgent(
        name="test_agent",
        description="Test agent",
        client=client,
        model="gpt-4",
        instructions="You are a helpful assistant.",
    )
    assistant: Dict[str, Any] = await agent.retrieve_assistant("asst_abc123")
    logging.info(assistant)


asyncio.run(example())

Modify (update) an assistant by its ID using the OpenAI API.

Parameters:

assistant_id (str) – The ID of the assistant to update.
name (Optional[str]) – New name for the assistant.
description (Optional[str]) – New description.
instructions (Optional[str]) – New instructions.
metadata (Optional[dict]) – New metadata.
model (Optional[str]) – New model.
reasoning_effort (Optional[str]) – New reasoning effort.
response_format (Optional[str]) – New response format.
temperature (Optional[float]) – New temperature.
tool_resources (Optional[dict]) – New tool resources.
tools (Optional[list]) – New tools.
top_p (Optional[float]) – New top_p value.
**kwargs – Additional keyword arguments.

Returns:

Dict[str, Any] – The updated assistant object.

Example

import asyncio
from typing import Dict, Any
from autogen_ext.agents.openai import OpenAIAgent
from openai import AsyncOpenAI
import logging


async def example() -> None:
    client = AsyncOpenAI()
    agent = OpenAIAgent(
        name="test_agent",
        description="Test agent",
        client=client,
        model="gpt-4",
        instructions="You are a helpful assistant.",
    )
    updated: Dict[str, Any] = await agent.modify_assistant(
        assistant_id="asst_123",
        instructions=(
            "You are an HR bot, and you have access to files to answer employee "
            "questions about company policies. Always response with info from either "
            "of the files."
        ),
        tools=[{"type": "file_search"}],
        tool_resources={"file_search": {"vector_store_ids": []}},
    )
    logging.info(updated)


asyncio.run(example())

async delete_assistant(assistant_id: str) → Dict[str, Any][source]#

Delete an assistant by its ID using the OpenAI API.

Parameters:: assistant_id (str) – The ID of the assistant to delete.
Returns:: Dict[str, Any] – The deletion status object.

Example

{“id”: “…”, “object”: “assistant.deleted”, “deleted”: true}

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.

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.

async on_upload_for_file_search(file_paths: str | Iterable[str], cancellation_token: CancellationToken) → None[source]#: 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.