evofabric.core.tool

evofabric.core.tool#

Base Tool#

class evofabric.core.tool.BaseTool(BaseComponent)[source]#

Basic tool class, used to encapsulate callable objects and manage its internal state and tool mode.

inherits from BaseComponent

Parameters:

func (Callable) – The function to be called can be a synchronous or asynchronous function.
exclude_params (List[str]) – List of parameter names excluded from display in tool mode
inner_state (Optional[ToolInnerState]) – The internal state of the tool. If the tool’s input parameters define inner_state: ToolInnerState, this parameter will be passed to the tool, which can read, modify, and update the state in real time.
name (Optional[str]) – Tool name, if not provided, use the function name
description (Optional[str]) – Tool description, if not provided, use the function docstring
tool_schema (Optional[dict]) – OpenAI-style tool mode

Usage example:

from evofabric.core.tool._tool_manager import ToolManager

tool_manager = ToolManager(tools=[])

async def add(a: float, b: float):
    """add two numbers"""
    return a + b

new_tool = BaseTool(
    name='add',
    description='add two float numbers.',
    func=add
)

tool_manager.add_callable_tools([new_tool])

__call__(**kwargs) → Any[source]#

Method for calling the tool.

Parameters:: kwargs – During tool invocation, if the tool excludes inner_state or stream_writer in exclude_params, the internal state or stream writer will be automatically passed.
Returns:: Tool execution result. If the tool is an asynchronous function, return an awaitable object.

model_post_init(context: Any, /)[source]#

Initialize BaseTool after Pydantic validation.

Parameters:: context (Any) – Context information for Pydantic validation

from_callable(func: Callable, name: str = None, description: str = None, tool_schema: dict | None = None, inner_state: ToolInnerState | None = None, exclude_params: List[str] = None)[source]#

Encapsulate the callable object as a BaseTool instance.

Parameters:

func (Callable) – The function to be called can be a synchronous or asynchronous function.
name (str) – Tool name, if not provided, use the function name
description (str) – Tool description, if not provided, use the function docstring
tool_schema (Optional[dict]) – OpenAI-style tool mode
inner_state (Optional[ToolInnerState]) – Internal state of the tool. If the tool’s input parameters define inner_state: ToolInnerState, this parameter will be passed to the tool
exclude_params (List[str]) – List of parameter names excluded from display in tool mode

Returns:

Encapsulated BaseTool instance

Return type:

BaseTool

get_tool_schema()[source]#

Get the OpenAI-style tool Schema description of a BaseTool instance.

Returns:: Tool Schema Description
Return type:: dict

dump_state()[source]#

Get the current internal state of the BaseTool instance.

Returns:: Internal state of the tool
Return type:: ToolInnerState

load_state(input_state: ToolInnerState)[source]#

Load the specified state to update the internal state of the BaseTool instance.

Parameters:: input_state (ToolInnerState) – Internal state to be loaded

Tool Manager#

class evofabric.core.tool.ToolManagerBase(ABC, BaseComponent)[source]#

Abstract base class for the tool management system, defining tool management-related interfaces. Inherits from abc.ABC and BaseComponent.

list_tools(**kwargs)[source]#

Return a list of tool modes in OpenAI style.

Raises:: NotImplementedError # Subclasses must implement this method

call_tools(tasks: List[ToolCall])[source]#

Execute tool calls based on the provided tool call task list.

Parameters:: tasks (List[ToolCall]) – Tool Call Task List
Raises:: NotImplementedError # Subclasses must implement this method

start()[source]#

The resources required to launch the tool management system.

Raises:: NotImplementedError # Subclasses must implement this method

stop()[source]#

Stop the resources that the tool management system relies on.

Raises:: NotImplementedError # Subclasses must implement this method

reset()[source]#

Reset the resources that the tool management system depends on.

Raises:: NotImplementedError # Subclasses must implement this method

save_state(save_path: str)[source]#

Save the internal state of the tools in the tool management system to the specified path.

Parameters:: save_path (str) – Path to save the tool’s internal state
Raises:: NotImplementedError # Subclasses must implement this method

load_state(load_path: str)[source]#

Load the tool’s internal state from the specified path into the corresponding tool in the tool management system.

Parameters:: load_path (str) – File path containing the tool’s internal state
Raises:: NotImplementedError # Subclasses must implement this method

class evofabric.core.tool.ToolManager(ToolManagerBase)[source]#

Implementation class of the conventional tool management system.

Inherits from ToolManagerBase, implements all its abstract methods, and provides tool management, invocation, import, deletion, update, and state management functionality.

Parameters:

tools (List[Union[Callable, Tuple[Callable, ToolInnerState], BaseTool]]) – Initial tool list, can include Python functions, BaseTool instances, or function tuples with internal state
timeout (int, 可选) – Tool execution timeout (seconds)
tool_controller (ToolController ，可选) – Tool Controller, used to control the activation state of the tools. Deactivated tools cannot be interacted with.

Usage example:

from evofabric.core.tool._tool_manager import ToolManager

async def MULTIPLY(a: int, b: int):
    """MULTIPLY two numbers."""
    return a * b

tool_manager = ToolManager(
    tools=[MULTIPLY],
)

model_post_init(context: Any, /)[source]#

After Pydantic validation, initialize ToolManager and add the tools to the system.

Parameters:: context (Any) – Pydantic Validation Context

list_tools()[source]#

List the tool modes of all tools in ToolManager.

Returns:: Tool Mode List
Return type:: List[dict]

call_tools(tasks: List[ToolCall])[source]#

Execute the tool call to complete the specified task.

Parameters:: tasks (List[ToolCall]) – Tool call task list, including tool names and parameters
Returns:: List of tool call results, each element containing call ID, result, and success flag, etc.
Return type:: List[ToolCallResult]

add_callable_tools(tools: List[Callable | BaseTool | Tuple[Callable, ToolInnerState]])[source]#

Add a Python function tool or a BaseTool instance to the system.

Parameters:: tools (List[Union[Callable, BaseTool, Tuple[Callable, ToolInnerState]]]) – List of Tools to Be Added

add_python_file_tools(file_paths: List[str], include_pattern_list: List[List[str]] = None, exclude_pattern_list: List[List[str]] = None)[source]#

Batch import qualifying function tools from Python files into the system.

Parameters:

file_paths (List[str]) – List of file paths containing Python function tools
include_pattern_list (List[List[str]]) – List of string patterns that must be included in function names within each file (length must match file_paths)
exclude_pattern_list (List[List[str]]) – List of string patterns that function names in each file cannot contain (length should match file_paths)

delete_tools(tool_names: List[str])[source]#

Delete the specified tool from the system.

Parameters:: tool_names (List[str]) – List of tool names to be deleted

update_tools(tools: List[BaseTool])[source]#

Update existing tool instances in the system.

Parameters:: tools (List[BaseTool]) – List of BaseTool instances to be updated

find_tools(tool_names: List[str])[source]#

Search for tools in the system.

Parameters:: tool_names (List[str]) – List of tool names to be looked up
Returns:: List of tool modes of the found tools
Return type:: List[dict]

save_state(save_path: str)[source]#

Save the internal state of all tools in the system to the specified path.

Parameters:: save_path (str) – Save Path

load_state(load_path: str)[source]#

Load the tool’s internal state from the specified path into the corresponding tool in the system.

Parameters:: load_path (str) – File path containing the tool’s internal state

dump_state(tool_name: str = None)[source]#

Retrieve the internal state of the tool in the system.

Parameters:: tool_name (str) – Specify the tool name, default is None, which means to get all tool statuses
Returns:: If tool_name is specified, return the status of that tool; otherwise, return the status of all tools
Return type:: dict

class evofabric.core.tool.McpToolManager(ToolManagerBase)[source]#

MCP (Model Context Protocol) Tool Manager, used to manage multiple MCP servers and their tools.

Attention:

The format of the tool name actually passed to the LLM is “server_name” + “_” + “tool_name”.

This naming convention supports the use of tools with the same name across different servers.

For example, if the server “math_server” has a tool “calculate”, its full tool name will be “math_server_calculate”.

Parameters:

server_links (Dict[str, McpServerLink]) – MCP server connection parameters dictionary, where the key is the server name and the value is the connection parameters.
timeout (int) – Tool execution timeout (seconds), default is 300
tool_controller (Optional[ToolController]) – A tool controller for managing the activation status of the tool, optional.
persistent_link (bool) – Whether to reuse the MCP connection flag. True means using the same connection for each interaction; False means creating a new connection each time.

async add_mcp_servers(mcp_server_links: Dict[str, McpServerLink])[source]#

Add a new MCP server to the manager.

Parameters:: mcp_server_links (Dict[str, McpServerLink]) – Dictionary from server name to McpServerLink configuration

async delete_mcp_servers(mcp_server_names: List[str])[source]#

Remove the MCP server from the manager and clean up the connections.

Parameters:: mcp_server_names (List[str]) – List of server names to be removed

async list_tools(server_name: str = None) → list[source]#

List available tools (optionally filtered by server name), filter status through the tool controller (if enabled).

Parameters:: server_name (str, optional) – Optional server name for filtering results
Returns:: Tool Mode List (status is filtered if the Tool Controller is enabled)
Return type:: list

async call_tools(tasks: List[ToolCall]) → List[ToolCallResult][source]#

Execute the provided tool call task list.

Parameters:: tasks (List[ToolCall]) – List of ToolCall objects to be executed
Returns:: A list of ToolCallResult objects containing execution results.
Return type:: List[ToolCallResult]

async list_prompts(server_name: str = None) → list[source]#

List available prompts (optionally filter by server name).

Parameters:: server_name (str, optional) – Optional server name for filtering results
Returns:: List of available prompts
Return type:: list

async list_resources(server_name: str = None) → list[source]#

List available resources (optionally filter by server name).

Parameters:: server_name (str, optional) – Optional server name for filtering results
Returns:: List of available resources
Return type:: list

async list_resource_templates(server_name: str = None) → list[source]#

List available resource templates (optionally filter by server name).

Parameters:: server_name (str, optional) – Optional server name for filtering results
Returns:: List of available resource templates
Return type:: list

async read_resource(resources: List[ResourceRequest]) → list[source]#

Read the specified resource from the MCP server.

Parameters:: resources (List[ResourceRequest]) – ResourceRequest object list
Returns:: Resource Response List
Return type:: list

async get_prompt(prompt_requests: List[PromptRequest]) → list[source]#

Retrieve the specified prompt from the MCP server.

Parameters:: prompt_requests (List[PromptRequest]) – PromptRequest object list
Returns:: Prompt Response List
Return type:: list

async set_tool_controller(controller: 'ToolController')[source]#

A tool controller for configuring the activation rules of management tools.

Parameters:: controller ('ToolController') – Tool controller instance with activate/deactivate rules

async get_tool_controller() → 'ToolController' | None[source]#

Get the current tool controller instance.

Returns:: The current ToolController instance returns None if not configured.
Return type:: Optional[‘ToolController’]

async connect(server_name: str = None)[source]#

Establish a connection to the MCP server.

Parameters:: server_name (str, optional) – Optional server name. If it is None, then connect to all servers.
Raises:: KeyError – If the specified server name does not exist

async disconnect(server_name: str = None)[source]#

Disconnect from the MCP server.

Parameters:: server_name (str, optional) – Optional server name. If None, disconnect all servers.
Raises:: KeyError – If the specified server name does not exist

async get_mcp_status() → Dict[str, bool][source]#

Get the connection status of all managed MCP servers.

Returns:: A dictionary that maps server names to connection status (True/False)
Return type:: Dict[str, bool]

async __aenter__() → McpToolManager[source]#

Enter the runtime context of McpToolManager. Establish connections to all configured MCP servers.

Returns:: McpToolManager instance for use in the context.
Return type:: McpToolManager

async __aexit__(exc_type, exc_val, exc_tb)[source]#

Exit the runtime context of McpToolManager. Automatically disconnect all MCP servers.

Parameters:

exc_type – Exception type when an exception occurs (otherwise None)
exc_val – The exception instance when an exception occurs (otherwise None)
exc_tb – The backtrace when an exception occurs (otherwise None)

class evofabric.core.tool.McpSessionController(BaseComponent)[source]#

A controller class that manages the lifecycle of MCP server connections and session operations. It is responsible for establishing connections, managing session state, and implementing graceful disconnection using an asynchronous context manager.

Parameters:

server_link (McpServerLink) – MCP Server Communication Link Object
server_name (str) – Name identifier of the connected MCP server
persistent_link (bool) – When set to True, exiting the context (aexit) will not disconnect the MCP server connection. Defaults to False.

property session#: Returns the current active client session (read-only property).

property is_connect#: Returns the connection status (True/False), for monitoring purposes.

async connect() → None[source]#: Start and wait for the MCP server connection to be established. Create a background task to maintain the session loop, blocking until the connection is fully established. Safely supports multiple calls; repeated calls will return early.

async disconnect() → None[source]#: Terminate the MCP connection. Cancel the session task, wait for cleanup to complete, and reset all connection states.

async _run_session_loop(ready_event: asyncio.Event) → None[source]#

Internal method: Run the main loop for session maintenance. After establishing the MCP session, wait for the connection ready event, enter a persistent waiting state until it is cancelled.

Parameters:: ready_event – An Event object used to notify that the connection is ready.

async __aenter__() → McpSessionController[source]#: Asynchronous context manager entry point. Automatically calls the connect() method to establish a connection, and returns its own instance.

async __aexit__(exc_type: Any, exc_val: Any, exc_tb: Any) → None[source]#: Asynchronous context manager exit. The disconnect() method is called to disconnect only when persistent_link is False.

Code Sandbox#

class evofabric.core.tool.CodeSandbox(BaseComponent)[source]#

Secure Python code sandbox, Docker-based, for executing Python code in an isolated environment.

Inherits from BaseComponent.

Parameters:: config (CodeExecDockerConfig) – Sandbox configuration, including image, container name, working directory, and other information

Usage example:

from evofabric.core.typing._tool import CodeExecDockerConfig
from evofabric.core.tool._code_sandbox import CodeSandbox

config = CodeExecDockerConfig(
    name="python_code_sandbox"
)
sandbox = CodeSandbox(config=config)
sandbox.start()

code1 = """
def fib(n):
    if n <= 2:
        return 1
    else:
        return fib(n-1) + fib(n-2)

res = fib(10)
print("10th fib number: ", res)
"""

result1 = sandbox.run_python(code1)
print("python result: ", result1)
sandbox.stop()

start()[source]#: Create and start the Docker container based on the configuration.

run_python(code: str)[source]#

Execute Python code in a sandbox container.

Parameters:: code (str) – Pending Python code
Returns:: Execution Result
Return type:: ExecResult

run_cmd(cmd: str)[source]#

Execute shell commands in the sandbox container.

Parameters:: cmd (str) – Pending shell command
Returns:: Execution Result
Return type:: ExecResult

stop()[source]#: Stop the Docker container. If the configuration’s auto_remove is True, the container will be automatically removed after it stops.

Tool Controller#

class evofabric.core.tool.ToolController(BaseComponent)[source]#

Tool Controller, used to manage the activation and deactivation status of tools. Inherits from :py:class`~evofabric.core.factory.BaseComponent`

Parameters:

default_mode (Literal['activate', 'deactivate']) – Default tool behavior, used when no rules match.
rules (list[Union[ToolControlPattern, dict]]) – A list of rules for controlling tool activation/deactivation, where rules are applied in order and the first matching rule determines the tool status.

Description of class attribute default values:

The default value for default_mode is “activate”. Optional values:

“activate”: Automatically activate tools that do not have matching rules (default value)
“deactivate”: Automatically disable tools that do not have matching rules.

The default value of rules is an empty dictionary. Each rule can be:

ToolControlPattern instance with the following attributes:

mode: ‘activate’ or ‘deactivate’

pattern: glob wildcard string

A dictionary with the following keys:

‘mode’: ‘activate’ or ‘deactivate’

‘pattern’: glob wildcard string

Note

Precautions when using McpToolManager:

The actual tool name format is [server_name]_[tool_name].
Wildcard pattern must be prefixed with server_name.
Example: To match all tools from the “math” server, use the pattern “math_*” (matches “math_calculator”, “math_grapher”, etc.)

Mode Example:

math_* → matches all tools from the “math” server (e.g., “math_calculator”, “math_grapher”)
text_* → matches all tools from the “text” server
math_calculator → only matches this specific tool
*_calculator → matches any tool ending with “*_calculator” (from any server)

check_tool_status(tool_name: str) → bool[source]#

Check whether the tool is activated according to the application rules.

Parameters:: tool_name (str) – Name of the tool to be checked
Returns:: If the tool is activated, return True, otherwise return False.
Return type:: bool

filter_tool_list(tool_list: List[Dict]) → List[Dict][source]#

Filter the tool list to only include activated tools.

Parameters:: tool_list (List[Dict]) – Tool Dictionary List
Returns:: List of activation tools
Return type:: List[Dict]

activate_tool(tool_name: str)[source]#

Activate the specific tool and assign it the highest priority. Remove any existing rules for this tool name.

Parameters:: tool_name (str) – The exact name of the tool to be activated.

deactivate_tool(tool_name: str)[source]#

Disable specific tool, assign highest priority. Remove any existing rules for this tool name.

Parameters:: tool_name (str) – The exact name of the tool to be disabled.

Utilities#

evofabric.core.tool.parse_callable_schema(function, name=None, description=None, include_long_description=True, include_var_positional=True, include_var_keyword=True, exclude_params=None)[source]#

Convert a Python callable object to a JSON schema for use with LLM tools. This function parses the function signature, docstring, and parameter default values to create a complete interface description compatible with the LLM tool calling format.

Supported function types include: regular functions, class methods, @classmethod, @staticmethod, partial objects of functions/class methods, lambda functions.

Parameters:

function (Callable) – Callable function to be converted to a pattern
name (Optional[str]) – Custom function name. If not provided, the function’s __name__ attribute will be used.
description (Optional[str]) – Custom description, if not provided, use the docstring summary.
include_long_description (bool) – Whether it contains the long description in the docstring.
include_var_positional (bool) – Whether it includes variable positional arguments (*args)
include_var_keyword (bool) – Does it contain keyword variable arguments (**kwargs)?
exclude_params (Optional[List[str]]) – List of parameter names to be excluded

Returns:

Tuple containing OpenAI-compatible function schema

Return type:

tuple

Return1:

A dictionary containing the complete function pattern.

Record Type 1:

dictionary

return to:

List of actually excluded parameter names.

Record type 2:

List[str]

evofabric.core.tool

Contents

evofabric.core.tool#

Base Tool#

Tool Manager#

Code Sandbox#

Tool Controller#

Utilities#