autogen_core.tools#

class Tool(*args, **kwargs)[source]#

Bases: Protocol

property name: str#

property description: str#

property schema: ToolSchema#

args_type() → Type[BaseModel][source]#

return_type() → Type[Any][source]#

state_type() → Type[BaseModel] | None[source]#

return_value_as_string(value: Any) → str[source]#

async run_json(args: Mapping[str, Any], cancellation_token: CancellationToken, call_id: str | None = None) → Any[source]#

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

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

class StreamTool(*args, **kwargs)[source]#

Bases: Tool, Protocol

run_json_stream(args: Mapping[str, Any], cancellation_token: CancellationToken, call_id: str | None = None) → AsyncGenerator[Any, None][source]#

class ToolSchema[source]#

Bases: TypedDict

parameters: NotRequired[ParametersSchema]#

name: str#

description: NotRequired[str]#

strict: NotRequired[bool]#

class ParametersSchema[source]#

Bases: TypedDict

type: str#

properties: Dict[str, Any]#

required: NotRequired[Sequence[str]]#

additionalProperties: NotRequired[bool]#

class BaseTool(args_type: Type[ArgsT], return_type: Type[ReturnT], name: str, description: str, strict: bool = False)[source]#

Bases: ABC, Tool, Generic[ArgsT, ReturnT], ComponentBase[BaseModel]

component_type: ClassVar[ComponentType] = 'tool'#: The logical type of the component.

property schema: ToolSchema#

property name: str#

property description: str#

args_type() → Type[BaseModel][source]#

return_type() → Type[Any][source]#

state_type() → Type[BaseModel] | None[source]#

return_value_as_string(value: Any) → str[source]#

abstract async run(args: ArgsT, cancellation_token: CancellationToken) → ReturnT[source]#

async run_json(args: Mapping[str, Any], cancellation_token: CancellationToken, call_id: str | None = None) → Any[source]#

Run the tool with the provided arguments in a dictionary.

Parameters:

args (Mapping[str, Any]) – The arguments to pass to the tool.
cancellation_token (CancellationToken) – A token to cancel the operation if needed.
call_id (str | None) – An optional identifier for the tool call, used for tracing.

Returns:

Any – The return value of the tool’s run method.

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

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

class BaseToolWithState(args_type: Type[ArgsT], return_type: Type[ReturnT], state_type: Type[StateT], name: str, description: str)[source]#

Bases: BaseTool[ArgsT, ReturnT], ABC, Generic[ArgsT, ReturnT, StateT], ComponentBase[BaseModel]

component_type: ClassVar[ComponentType] = 'tool'#: The logical type of the component.

abstract save_state() → StateT[source]#

abstract load_state(state: StateT) → None[source]#

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

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

class BaseStreamTool(args_type: Type[ArgsT], return_type: Type[ReturnT], name: str, description: str, strict: bool = False)[source]#

Bases: BaseTool[ArgsT, ReturnT], StreamTool, ABC, Generic[ArgsT, StreamT, ReturnT], ComponentBase[BaseModel]

component_type: ClassVar[ComponentType] = 'tool'#: The logical type of the component.

abstract run_stream(args: ArgsT, cancellation_token: CancellationToken) → AsyncGenerator[StreamT | ReturnT, None][source]#: Run the tool with the provided arguments and return a stream of data and end with the final return value.

async run_json_stream(args: Mapping[str, Any], cancellation_token: CancellationToken, call_id: str | None = None) → AsyncGenerator[StreamT | ReturnT, None][source]#

Run the tool with the provided arguments in a dictionary and return a stream of data from the tool’s run_stream() method and end with the final return value.

Parameters:

args (Mapping[str, Any]) – The arguments to pass to the tool.
cancellation_token (CancellationToken) – A token to cancel the operation if needed.
call_id (str | None) – An optional identifier for the tool call, used for tracing.

Returns:

AsyncGenerator[StreamT | ReturnT, None] – A generator yielding results from the tool’s run_stream() method.

class FunctionTool(func: Callable[[...], Any], description: str, name: str | None = None, global_imports: Sequence[str | ImportFromModule | Alias] = [], strict: bool = False)[source]#

Bases: BaseTool[BaseModel, BaseModel], Component[FunctionToolConfig]

Create custom tools by wrapping standard Python functions.

FunctionTool offers an interface for executing Python functions either asynchronously or synchronously. Each function must include type annotations for all parameters and its return type. These annotations enable FunctionTool to generate a schema necessary for input validation, serialization, and for informing the LLM about expected parameters. When the LLM prepares a function call, it leverages this schema to generate arguments that align with the function’s specifications.

Note

It is the user’s responsibility to verify that the tool’s output type matches the expected type.

Parameters:

func (Callable[..., ReturnT | Awaitable[ReturnT]]) – The function to wrap and expose as a tool.
description (str) – A description to inform the model of the function’s purpose, specifying what it does and the context in which it should be called.
name (str, optional) – An optional custom name for the tool. Defaults to the function’s original name if not provided.
strict (bool, optional) – If set to True, the tool schema will only contain arguments that are explicitly defined in the function signature, and no default values will be allowed. Defaults to False. This is required to be set to True when used with models in structured output mode.

Example

import random
from autogen_core import CancellationToken
from autogen_core.tools import FunctionTool
from typing_extensions import Annotated
import asyncio


async def get_stock_price(ticker: str, date: Annotated[str, "Date in YYYY/MM/DD"]) -> float:
    # Simulates a stock price retrieval by returning a random float within a specified range.
    return random.uniform(10, 200)


async def example():
    # Initialize a FunctionTool instance for retrieving stock prices.
    stock_price_tool = FunctionTool(get_stock_price, description="Fetch the stock price for a given ticker.")

    # Execute the tool with cancellation support.
    cancellation_token = CancellationToken()
    result = await stock_price_tool.run_json({"ticker": "AAPL", "date": "2021/01/01"}, cancellation_token)

    # Output the result as a formatted string.
    print(stock_price_tool.return_value_as_string(result))


asyncio.run(example())

component_provider_override: ClassVar[str | None] = 'autogen_core.tools.FunctionTool'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

component_config_schema#: alias of FunctionToolConfig

async run(args: BaseModel, cancellation_token: CancellationToken) → Any[source]#

_to_config() → FunctionToolConfig[source]#

Dump the configuration that would be requite to create a new instance of a component matching the configuration of this instance.

Returns:: T – The configuration of the component.

classmethod _from_config(config: FunctionToolConfig) → Self[source]#

Create a new instance of the component from a configuration object.

Parameters:: config (T) – The configuration object.
Returns:: Self – The new instance of the component.

class Workbench[source]#

Bases: ABC, ComponentBase[BaseModel]

A workbench is a component that provides a set of tools that may share resources and state.

A workbench is responsible for managing the lifecycle of the tools and providing a single interface to call them. The tools provided by the workbench may be dynamic and their availabilities may change after each tool execution.

A workbench can be started by calling the start() method and stopped by calling the stop() method. It can also be used as an asynchronous context manager, which will automatically start and stop the workbench when entering and exiting the context.

component_type: ClassVar[ComponentType] = 'workbench'#: The logical type of the component.

abstract async list_tools() → List[ToolSchema][source]#

List the currently available tools in the workbench as ToolSchema objects.

The list of tools may be dynamic, and their content may change after tool execution.

abstract async call_tool(name: str, arguments: Mapping[str, Any] | None = None, cancellation_token: CancellationToken | None = None, call_id: str | None = None) → ToolResult[source]#

Call a tool in the workbench.

Parameters:

name (str) – The name of the tool to call.
arguments (Mapping[str, Any] | None) – The arguments to pass to the tool. If None, the tool will be called with no arguments.
cancellation_token (CancellationToken | None) – An optional cancellation token to cancel the tool execution.
call_id (str | None) – An optional identifier for the tool call, used for tracing.

Returns:

ToolResult – The result of the tool execution.

abstract async start() → None[source]#

Start the workbench and initialize any resources.

This method should be called before using the workbench.

abstract async stop() → None[source]#

Stop the workbench and release any resources.

This method should be called when the workbench is no longer needed.

abstract async reset() → None[source]#: Reset the workbench to its initialized, started state.

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

Save the state of the workbench.

This method should be called to persist the state of the workbench.

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

Load the state of the workbench.

Parameters:: state (Mapping[str, Any]) – The state to load into the workbench.

pydantic model ToolResult[source]#

Bases: BaseModel

A result of a tool execution by a workbench.

Show JSON schema

{
   "title": "ToolResult",
   "description": "A result of a tool execution by a workbench.",
   "type": "object",
   "properties": {
      "type": {
         "const": "ToolResult",
         "default": "ToolResult",
         "title": "Type",
         "type": "string"
      },
      "name": {
         "title": "Name",
         "type": "string"
      },
      "result": {
         "items": {
            "discriminator": {
               "mapping": {
                  "ImageResultContent": "#/$defs/ImageResultContent",
                  "TextResultContent": "#/$defs/TextResultContent"
               },
               "propertyName": "type"
            },
            "oneOf": [
               {
                  "$ref": "#/$defs/TextResultContent"
               },
               {
                  "$ref": "#/$defs/ImageResultContent"
               }
            ]
         },
         "title": "Result",
         "type": "array"
      },
      "is_error": {
         "default": false,
         "title": "Is Error",
         "type": "boolean"
      }
   },
   "$defs": {
      "ImageResultContent": {
         "description": "Image result content of a tool execution.",
         "properties": {
            "type": {
               "const": "ImageResultContent",
               "default": "ImageResultContent",
               "title": "Type",
               "type": "string"
            },
            "content": {
               "title": "Content"
            }
         },
         "required": [
            "content"
         ],
         "title": "ImageResultContent",
         "type": "object"
      },
      "TextResultContent": {
         "description": "Text result content of a tool execution.",
         "properties": {
            "type": {
               "const": "TextResultContent",
               "default": "TextResultContent",
               "title": "Type",
               "type": "string"
            },
            "content": {
               "title": "Content",
               "type": "string"
            }
         },
         "required": [
            "content"
         ],
         "title": "TextResultContent",
         "type": "object"
      }
   },
   "required": [
      "name",
      "result"
   ]
}

Fields:

is_error (bool)
name (str)
result (List[autogen_core.tools._workbench.TextResultContent | autogen_core.tools._workbench.ImageResultContent])
type (Literal['ToolResult'])

field type: Literal['ToolResult'] = 'ToolResult'#

field name: str [Required]#: The name of the tool that was executed.

field result: List[Annotated[TextResultContent | ImageResultContent, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] [Required]#: The result of the tool execution.

field is_error: bool = False#: Whether the tool execution resulted in an error.

to_text(replace_image: str | None = None) → str[source]#

Convert the result to a text string.

Parameters:: replace_image (str | None) – The string to replace the image content with. If None, the image content will be included in the text as base64 string.
Returns:: str – The text representation of the result.

pydantic model TextResultContent[source]#

Bases: BaseModel

Text result content of a tool execution.

Show JSON schema

{
   "title": "TextResultContent",
   "description": "Text result content of a tool execution.",
   "type": "object",
   "properties": {
      "type": {
         "const": "TextResultContent",
         "default": "TextResultContent",
         "title": "Type",
         "type": "string"
      },
      "content": {
         "title": "Content",
         "type": "string"
      }
   },
   "required": [
      "content"
   ]
}

Fields:

content (str)
type (Literal['TextResultContent'])

field type: Literal['TextResultContent'] = 'TextResultContent'#

field content: str [Required]#: The text content of the result.

pydantic model ImageResultContent[source]#

Bases: BaseModel

Image result content of a tool execution.

Show JSON schema

{
   "title": "ImageResultContent",
   "description": "Image result content of a tool execution.",
   "type": "object",
   "properties": {
      "type": {
         "const": "ImageResultContent",
         "default": "ImageResultContent",
         "title": "Type",
         "type": "string"
      },
      "content": {
         "title": "Content"
      }
   },
   "required": [
      "content"
   ]
}

Fields:

content (autogen_core._image.Image)
type (Literal['ImageResultContent'])

field type: Literal['ImageResultContent'] = 'ImageResultContent'#

field content: Image [Required]#: The image content of the result.

class StaticWorkbench(tools: List[BaseTool[Any, Any]], tool_overrides: Dict[str, ToolOverride] | None = None)[source]#

Bases: Workbench, Component[StaticWorkbenchConfig]

A workbench that provides a static set of tools that do not change after each tool execution.

Parameters:

tools (List[BaseTool[Any, Any]]) – A list of tools to be included in the workbench. The tools should be subclasses of BaseTool.
tool_overrides (Optional[Dict[str, ToolOverride]]) – Optional mapping of original tool names to override configurations for name and/or description. This allows customizing how tools appear to consumers while maintaining the underlying tool functionality.

component_provider_override: ClassVar[str | None] = 'autogen_core.tools.StaticWorkbench'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

component_config_schema#: alias of StaticWorkbenchConfig

async list_tools() → List[ToolSchema][source]#

List the currently available tools in the workbench as ToolSchema objects.

The list of tools may be dynamic, and their content may change after tool execution.

async call_tool(name: str, arguments: Mapping[str, Any] | None = None, cancellation_token: CancellationToken | None = None, call_id: str | None = None) → ToolResult[source]#

Call a tool in the workbench.

Parameters:

name (str) – The name of the tool to call.
arguments (Mapping[str, Any] | None) – The arguments to pass to the tool. If None, the tool will be called with no arguments.
cancellation_token (CancellationToken | None) – An optional cancellation token to cancel the tool execution.
call_id (str | None) – An optional identifier for the tool call, used for tracing.

Returns:

ToolResult – The result of the tool execution.

async start() → None[source]#

Start the workbench and initialize any resources.

This method should be called before using the workbench.

async stop() → None[source]#

Stop the workbench and release any resources.

This method should be called when the workbench is no longer needed.

async reset() → None[source]#: Reset the workbench to its initialized, started state.

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

Save the state of the workbench.

This method should be called to persist the state of the workbench.

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

Load the state of the workbench.

Parameters:: state (Mapping[str, Any]) – The state to load into the workbench.

_to_config() → StaticWorkbenchConfig[source]#

Dump the configuration that would be requite to create a new instance of a component matching the configuration of this instance.

Returns:: T – The configuration of the component.

classmethod _from_config(config: StaticWorkbenchConfig) → Self[source]#

Create a new instance of the component from a configuration object.

Parameters:: config (T) – The configuration object.
Returns:: Self – The new instance of the component.

class StaticStreamWorkbench(tools: List[BaseTool[Any, Any]], tool_overrides: Dict[str, ToolOverride] | None = None)[source]#

Bases: StaticWorkbench, StreamWorkbench

A workbench that provides a static set of tools that do not change after each tool execution, and supports streaming results.

component_provider_override: ClassVar[str | None] = 'autogen_core.tools.StaticStreamWorkbench'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

async call_tool_stream(name: str, arguments: Mapping[str, Any] | None = None, cancellation_token: CancellationToken | None = None, call_id: str | None = None) → AsyncGenerator[Any | ToolResult, None][source]#

Call a tool in the workbench and return a stream of results.

Parameters:

name (str) – The name of the tool to call.
arguments (Mapping[str, Any] | None) – The arguments to pass to the tool If None, the tool will be called with no arguments.
cancellation_token (CancellationToken | None) – An optional cancellation token to cancel the tool execution.
call_id (str | None) – An optional identifier for the tool call, used for tracing.

pydantic model ToolOverride[source]#

Bases: BaseModel

Override configuration for a tool’s name and/or description.

Show JSON schema

{
   "title": "ToolOverride",
   "description": "Override configuration for a tool's name and/or description.",
   "type": "object",
   "properties": {
      "name": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Name"
      },
      "description": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Description"
      }
   }
}

Fields:

description (str | None)
name (str | None)

field name: str | None = None#

field description: str | None = None#