OpenAIAssistantAgent Class

OpenAI Assistant Agent class, providing the ability to interact with OpenAI's chat-based Assistants. Offers conversation-level streaming, message handling, and plugin-based expansions.

Constructor

OpenAIAssistantAgent(
  *,
  arguments: KernelArguments | None = None,
  client: AsyncOpenAI,
  definition: Assistant,
  kernel: Kernel | None = None,
  plugins: list[KernelPlugin | object] | dict[str, KernelPlugin | object] | None = None,
  polling_options: RunPollingOptions | None = None,
  prompt_template_config: PromptTemplateConfig | None = None,
  **kwargs: Any
)

Keyword-Only Parameters

Name	Description
arguments	KernelArguments to pass for initialization or overrides.
client Required	The OpenAI Async client instance for chat endpoints.
definition Required	The assistant definition (name, description, instructions).
kernel	The Kernel instance for plugin usage.
plugins	A collection of plugin objects or a dict of plugins.
polling_options	Options controlling how the agent polls for results or intermediate steps.
prompt_template_config	Optional template config that can override the agent instructions.
kwargs	Additional dynamic keyword args stored in agent state.

Methods

setup_resources	Create an AsyncOpenAI client from provided or environment-based settings, returning a tuple of (client, model_name).
configure_code_interpreter_tool	Generate the tool + resources needed to enable the "code_interpreter" functionality.
configure_file_search_tool	Generate the tool + resources needed to enable "file_search" functionality.
configure_response_format	Formulate a structured response format configuration for JSON-based or schema-based outputs, bridging to the openai beta's param format.
get_channel_keys	Return channel keys that uniquely identify the agent's OpenAI channel usage: including agent ID, name, and base URL.
create_channel	Create a new or reuse an existing conversation thread, returning a channel object that binds to the chat session on the service.
add_chat_message	Add a user or system message to the specified thread by ID.
get_response	Retrieve a final ChatMessageContent from the agent after passing messages or instructions. Streams internally, but returns only the last visible message.
invoke	Invoke the agent for a single pass, returning each visible message as it's created. The final message is typically the agent's response.
invoke_stream	Invoke the agent in streaming mode, emitting partial messages as they are formed. Once complete, optionally calls a provided callback with the fully constructed conversation.

setup_resources

Create an AsyncOpenAI client from provided or environment-based settings, returning a tuple of (client, model_name).

static setup_resources(
  *,
  ai_model_id: str | None = None,
  api_key: str | None = None,
  org_id: str | None = None,
  env_file_path: str | None = None,
  env_file_encoding: str | None = None,
  default_headers: dict[str, str] | None = None,
  **kwargs: Any
) -> tuple[AsyncOpenAI, str]

Parameters

Name	Description
ai_model_id	The ID of the OpenAI chat model to use.
api_key	The OpenAI API key.
org_id	The OpenAI organization ID.
env_file_path	The path to a .env file for environment variables.
env_file_encoding	Encoding used to load the .env file. Defaults to utf-8.
default_headers	Default headers to add to the client (e.g. for custom user-agents).
kwargs	Additional advanced arguments for the client.

Returns

Type	Description
	A tuple of (AsyncOpenAI client, chat_model_id).

configure_code_interpreter_tool

Generate the tool + resources needed to enable the "code_interpreter" functionality.

static configure_code_interpreter_tool(
  file_ids: str | list[str] | None = None,
  **kwargs: Any
) -> tuple[list[CodeInterpreterToolParam], ToolResources]

configure_file_search_tool

Generate the tool + resources needed to enable "file_search" functionality.

static configure_file_search_tool(
  vector_store_ids: str | list[str],
  **kwargs: Any
) -> tuple[list[FileSearchToolParam], ToolResources]

configure_response_format

Formulate a structured response format configuration for JSON-based or schema-based outputs, bridging to the openai beta's param format.

static configure_response_format(
  response_format: dict[Literal["type"], Literal["text","json_object"]] 
    | dict[str, Any] 
    | type[BaseModel] 
    | type 
    | AssistantResponseFormatOptionParam 
    | None = None
) -> AssistantResponseFormatOptionParam | None

get_channel_keys

Return channel keys that uniquely identify the agent's OpenAI channel usage: including agent ID, name, and base URL.

get_channel_keys() -> Iterable[str]

create_channel

Create a new or reuse an existing conversation thread, returning a channel object that binds to the chat session on the service.

async create_channel(thread_id: str | None = None) -> AgentChannel

Parameters

Name	Description
thread_id	The optional ID of an existing thread. If None, a new thread is created. Default value: None

Returns

Type	Description
	The newly created or existing OpenAIAssistantChannel instance.

add_chat_message

Add a user or system message to the specified thread by ID.

async add_chat_message(thread_id: str, message: str | ChatMessageContent, **kwargs: Any) -> Message | None

Parameters

Name	Description
thread_id Required	The ID of the thread to add a message to.
message Required	The message content or string.

Returns

Type	Description
	The newly created message object or None if creation failed.

get_response

Retrieve a final ChatMessageContent from the agent after passing messages or instructions. Streams internally, but returns only the last visible message.

async get_response(
  *,
  messages: str | ChatMessageContent | list[str | ChatMessageContent] | None = None,
  thread: AgentThread | None = None,
  arguments: KernelArguments | None = None,
  additional_instructions: str | None = None,
  additional_messages: list[ChatMessageContent] | None = None,
  instructions_override: str | None = None,
  kernel: Kernel | None = None,
  max_completion_tokens: int | None = None,
  max_prompt_tokens: int | None = None,
  metadata: dict[str, str] | None = None,
  model: str | None = None,
  parallel_tool_calls: bool | None = None,
  reasoning_effort: Literal["low","medium","high"] | None = None,
  response_format: AssistantResponseFormatOptionParam | None = None,
  tools: list[AssistantToolParam] | None = None,
  temperature: float | None = None,
  top_p: float | None = None,
  truncation_strategy: TruncationStrategy | None = None,
  **kwargs: Any
) -> AgentResponseItem[ChatMessageContent]

invoke

Invoke the agent for a single pass, returning each visible message as it's created. The final message is typically the agent's response.

async invoke(
  *,
  messages: str | ChatMessageContent | list[str | ChatMessageContent] | None = None,
  thread: AgentThread | None = None,
  arguments: KernelArguments | None = None,
  additional_instructions: str | None = None,
  additional_messages: list[ChatMessageContent] | None = None,
  instructions_override: str | None = None,
  kernel: Kernel | None = None,
  max_completion_tokens: int | None = None,
  max_prompt_tokens: int | None = None,
  metadata: dict[str, str] | None = None,
  model: str | None = None,
  parallel_tool_calls: bool | None = None,
  reasoning_effort: Literal["low","medium","high"] | None = None,
  response_format: AssistantResponseFormatOptionParam | None = None,
  tools: list[AssistantToolParam] | None = None,
  temperature: float | None = None,
  top_p: float | None = None,
  truncation_strategy: TruncationStrategy | None = None,
  **kwargs: Any
) -> AsyncIterable[AgentResponseItem[ChatMessageContent]]

invoke_stream

Invoke the agent in streaming mode, emitting partial messages as they are formed. Once complete, optionally calls a provided callback with the fully constructed conversation.

async invoke_stream(
  *,
  messages: str | ChatMessageContent | list[str | ChatMessageContent] | None = None,
  thread: AgentThread | None = None,
  additional_instructions: str | None = None,
  additional_messages: list[ChatMessageContent] | None = None,
  arguments: KernelArguments | None = None,
  instructions_override: str | None = None,
  kernel: Kernel | None = None,
  max_completion_tokens: int | None = None,
  max_prompt_tokens: int | None = None,
  metadata: dict[str, str] | None = None,
  model: str | None = None,
  on_complete: Callable[[ChatHistory], None] | None = None,
  parallel_tool_calls: bool | None = None,
  reasoning_effort: Literal["low","medium","high"] | None = None,
  response_format: AssistantResponseFormatOptionParam | None = None,
  tools: list[AssistantToolParam] | None = None,
  temperature: float | None = None,
  top_p: float | None = None,
  truncation_strategy: TruncationStrategy | None = None,
  **kwargs: Any
) -> AsyncIterable[AgentResponseItem[StreamingChatMessageContent]]

Attributes

client

The AsyncOpenAI client bound to this agent.

client: AsyncOpenAI

definition

The raw assistant definition object, containing the name, description, etc.

definition: Assistant

plugins

Optional list of plugin objects or references used by the agent.

plugins: list[Any]

polling_options

RunPollingOptions controlling the agent's approach to waiting for completions, tool calls, etc.

polling_options: RunPollingOptions

channel_type

The default AgentChannel used, set to OpenAIAssistantChannel.

channel_type: ClassVar[type[AgentChannel]] = OpenAIAssistantChannel

Share via

OpenAIAssistantAgent Class

Constructor

Keyword-Only Parameters

Methods

setup_resources

Parameters

Returns

configure_code_interpreter_tool

configure_file_search_tool

configure_response_format

get_channel_keys

create_channel

Parameters

Returns

add_chat_message

Parameters

Returns

get_response

invoke

invoke_stream

Attributes

client

definition

plugins

polling_options

channel_type