evofabric.core.tool

evofabric.core.tool#

Base Tool#

class evofabric.core.tool.BaseTool(BaseComponent)[源代码]#

基础工具类，用于封装可调用对象并管理其内部状态与工具模式。

继承自 BaseComponent

参数:

func (Callable) -- 要调用的函数，可以是同步或异步函数
exclude_params (List[str]) -- 在工具模式中排除显示的参数名称列表
inner_state (Optional[ToolInnerState]) -- 工具的内部状态。如果工具的输入参数定义了 inner_state: ToolInnerState，此参数会被传递给工具，工具可读取、修改并实时更新该状态
name (Optional[str]) -- 工具的名称，未提供则使用函数名称
description (Optional[str]) -- 工具的描述，未提供则使用函数文档字符串
tool_schema (Optional[dict]) -- OpenAI 风格的工具模式

使用示例：

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[源代码]#

调用该工具的方法。

参数:: kwargs -- 工具调用时的参数，如果工具在 exclude_params 中排除了 inner_state 或 stream_writer，会自动传入内部状态或流写入器。
返回:: 工具执行结果，如果工具为异步函数，则返回 awaitable 对象

model_post_init(context: Any, /)[源代码]#

在 Pydantic 验证完成后初始化 BaseTool。

参数:: context (Any) -- Pydantic 验证的上下文信息

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)[源代码]#

将可调用对象封装为 BaseTool 实例。

参数:

func (Callable) -- 要调用的函数，可以是同步或异步函数
name (str) -- 工具的名称，未提供则使用函数名称
description (str) -- 工具的描述，未提供则使用函数文档字符串
tool_schema (Optional[dict]) -- OpenAI 风格的工具模式
inner_state (Optional[ToolInnerState]) -- 工具的内部状态。如果工具的输入参数定义了 inner_state: ToolInnerState，此参数会被传递给工具
exclude_params (List[str]) -- 在工具模式中排除显示的参数名称列表

返回:

封装后的 BaseTool 实例

返回类型:

BaseTool

get_tool_schema()[源代码]#

获取 BaseTool 实例的 OpenAI 风格工具 Schema 描述。

返回:: 工具 Schema 描述
返回类型:: dict

dump_state()[源代码]#

获取 BaseTool 实例的当前内部状态。

返回:: 工具的内部状态
返回类型:: ToolInnerState

load_state(input_state: ToolInnerState)[源代码]#

加载指定状态以更新 BaseTool 实例的内部状态。

参数:: input_state (ToolInnerState) -- 要加载的内部状态

Tool Manager#

class evofabric.core.tool.ToolManagerBase(ABC, BaseComponent)[源代码]#

工具管理系统的基础抽象类，定义工具管理相关接口。继承自 abc.ABC 和 BaseComponent。

list_tools(**kwargs)[源代码]#

以 OpenAI 风格返回工具模式列表。

Raises:: NotImplementedError # 子类必须实现此方法

call_tools(tasks: List[ToolCall])[源代码]#

根据提供的工具调用任务列表执行工具调用。

参数:: tasks (List[ToolCall]) -- 工具调用任务列表
Raises:: NotImplementedError # 子类必须实现此方法

start()[源代码]#

启动工具管理系统所依赖的资源。

Raises:: NotImplementedError # 子类必须实现此方法

stop()[源代码]#

停止工具管理系统所依赖的资源。

Raises:: NotImplementedError # 子类必须实现此方法

reset()[源代码]#

重置工具管理系统所依赖的资源。

Raises:: NotImplementedError # 子类必须实现此方法

save_state(save_path: str)[源代码]#

将工具管理系统中工具的内部状态保存到指定路径。

参数:: save_path (str) -- 保存工具内部状态的路径
Raises:: NotImplementedError # 子类必须实现此方法

load_state(load_path: str)[源代码]#

从指定路径加载工具内部状态到工具管理系统中对应的工具。

参数:: load_path (str) -- 包含工具内部状态的文件路径
Raises:: NotImplementedError # 子类必须实现此方法

class evofabric.core.tool.ToolManager(ToolManagerBase)[源代码]#

常规工具管理系统的实现类。

继承自 ToolManagerBase，实现其所有抽象方法，提供工具管理、调用、导入、删除、更新及状态管理功能。

参数:

tools (List[Union[Callable, Tuple[Callable, ToolInnerState], BaseTool]]) -- 初始工具列表，可包含 Python 函数、BaseTool 实例或带内部状态的函数元组
timeout (int, 可选) -- 工具执行的超时时间（秒）
tool_controller (ToolController ，可选) -- 工具控制器，用来控制工具的激活状态，未激活的工具无法交互。

使用示例：

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, /)[源代码]#

在 Pydantic 验证后初始化 ToolManager，并将工具添加到系统中。

参数:: context (Any) -- Pydantic 验证上下文

list_tools()[源代码]#

列出 ToolManager 中所有工具的工具模式。

返回:: 工具模式列表
返回类型:: List[dict]

call_tools(tasks: List[ToolCall])[源代码]#

执行工具调用以完成指定任务。

参数:: tasks (List[ToolCall]) -- 工具调用任务列表，包括工具名称及参数
返回:: 工具调用结果列表，每个元素包含调用 ID、结果及成功标志等信息
返回类型:: List[ToolCallResult]

add_callable_tools(tools: List[Callable | BaseTool | Tuple[Callable, ToolInnerState]])[源代码]#

向系统中添加 Python 函数工具或 BaseTool 实例。

参数:: tools (List[Union[Callable, BaseTool, Tuple[Callable, ToolInnerState]]]) -- 待添加的工具列表

add_python_file_tools(file_paths: List[str], include_pattern_list: List[List[str]] = None, exclude_pattern_list: List[List[str]] = None)[源代码]#

从 Python 文件批量导入符合条件的函数工具到系统中。

参数:

file_paths (List[str]) -- 包含 Python 函数工具的文件路径列表
include_pattern_list (List[List[str]]) -- 每个文件中函数名必须包含的字符串模式列表（长度应与 file_paths 一致）
exclude_pattern_list (List[List[str]]) -- 每个文件中函数名不能包含的字符串模式列表（长度应与 file_paths 一致）

delete_tools(tool_names: List[str])[源代码]#

从系统中删除指定工具。

参数:: tool_names (List[str]) -- 待删除的工具名称列表

update_tools(tools: List[BaseTool])[源代码]#

更新系统中已有的工具实例。

参数:: tools (List[BaseTool]) -- 待更新的 BaseTool 实例列表

find_tools(tool_names: List[str])[源代码]#

查找系统中的工具。

参数:: tool_names (List[str]) -- 待查找的工具名称列表
返回:: 找到的工具的工具模式列表
返回类型:: List[dict]

save_state(save_path: str)[源代码]#

保存系统中所有工具的内部状态到指定路径。

参数:: save_path (str) -- 保存路径

load_state(load_path: str)[源代码]#

从指定路径加载工具内部状态到系统中对应的工具。

参数:: load_path (str) -- 包含工具内部状态的文件路径

dump_state(tool_name: str = None)[源代码]#

获取系统中工具的内部状态。

参数:: tool_name (str) -- 指定工具名称，默认为 None，表示获取所有工具状态
返回:: 如果指定了 tool_name，则返回该工具的状态；否则返回所有工具的状态
返回类型:: dict

class evofabric.core.tool.McpToolManager(ToolManagerBase)[源代码]#

MCP (Model Context Protocol) 工具管理器，用于管理多个 MCP 服务器及其工具。

注意：

实际传递给 LLM 的工具名称格式为 "server_name" + "_" + "tool_name"

此命名约定支持在不同服务器中使用同名的工具

例如，如果服务器 "math_server" 有工具 "calculate"，其完整工具名将为 "math_server_calculate"

参数:

server_links (Dict[str, McpServerLink]) -- MCP 服务器连接参数字典，键为服务器名称，值为连接参数
timeout (int) -- 工具执行超时时间（秒），默认为300
tool_controller (Optional[ToolController]) -- 用于管理工作具激活状态的工具控制器，可选
persistent_link (bool) -- 是否重用 MCP 连接标志。True 表示每次交互使用相同连接；False 表示每次创建新连接

async add_mcp_servers(mcp_server_links: Dict[str, McpServerLink])[源代码]#

向管理器添加新的 MCP 服务器。

参数:: mcp_server_links (Dict[str, McpServerLink]) -- 服务器名称到 McpServerLink 配置的字典

async delete_mcp_servers(mcp_server_names: List[str])[源代码]#

从管理器中移除 MCP 服务器并清理连接。

参数:: mcp_server_names (List[str]) -- 要移除的服务器名称列表

async list_tools(server_name: str = None) → list[源代码]#

列出可用工具（可选按服务器名筛选），通过工具控制器过滤状态（如果已启用）。

参数:: server_name (str, optional) -- 可选的服务器名称用于筛选结果
返回:: 工具模式列表（如果启用了工具控制器则会过滤状态）
返回类型:: list

async call_tools(tasks: List[ToolCall]) → List[ToolCallResult][源代码]#

执行提供的工具调用任务列表。

参数:: tasks (List[ToolCall]) -- 要执行的 ToolCall 对象列表
返回:: 包含执行结果的 ToolCallResult 对象列表
返回类型:: List[ToolCallResult]

async list_prompts(server_name: str = None) → list[源代码]#

列出可用提示（可选按服务器名筛选）。

参数:: server_name (str, optional) -- 可选的服务器名称用于筛选结果
返回:: 可用提示的列表
返回类型:: list

async list_resources(server_name: str = None) → list[源代码]#

列出可用资源（可选按服务器名筛选）。

参数:: server_name (str, optional) -- 可选的服务器名称用于筛选结果
返回:: 可用资源的列表
返回类型:: list

async list_resource_templates(server_name: str = None) → list[源代码]#

列出可用资源模板（可选按服务器名筛选）。

参数:: server_name (str, optional) -- 可选的服务器名称用于筛选结果
返回:: 可用资源模板的列表
返回类型:: list

async read_resource(resources: List[ResourceRequest]) → list[源代码]#

从 MCP 服务器读取指定的资源。

参数:: resources (List[ResourceRequest]) -- ResourceRequest 对象列表
返回:: 资源响应列表
返回类型:: list

async get_prompt(prompt_requests: List[PromptRequest]) → list[源代码]#

从 MCP 服务器获取指定的提示。

参数:: prompt_requests (List[PromptRequest]) -- PromptRequest 对象列表
返回:: 提示响应列表
返回类型:: list

async set_tool_controller(controller: 'ToolController')[源代码]#

设置管理工作具激活规则的工具控制器。

参数:: controller ('ToolController') -- 包含激活/停用规则的工具控制器实例

async get_tool_controller() → 'ToolController' | None[源代码]#

获取当前工具控制器实例。

返回:: 当前的 ToolController 实例，如果未配置则返回 None
返回类型:: Optional['ToolController']

async connect(server_name: str = None)[源代码]#

建立 MCP 服务器的连接。

参数:: server_name (str, optional) -- 可选的服务器名称。如果为 None，则连接所有服务器
抛出:: KeyError -- 如果指定的服务器名称不存在

async disconnect(server_name: str = None)[源代码]#

断开 MCP 服务器的连接。

参数:: server_name (str, optional) -- 可选的服务器名称。如果为 None，则断开所有服务器
抛出:: KeyError -- 如果指定的服务器名称不存在

async get_mcp_status() → Dict[str, bool][源代码]#

获取所有管理的 MCP 服务器的连接状态。

返回:: 映射服务器名称到连接状态（True/False）的字典
返回类型:: Dict[str, bool]

async __aenter__() → McpToolManager[源代码]#

进入 McpToolManager 的运行时上下文。建立所有配置的 MCP 服务器的连接。

返回:: 供上下文内使用的 McpToolManager 实例
返回类型:: McpToolManager

async __aexit__(exc_type, exc_val, exc_tb)[源代码]#

退出 McpToolManager 的运行时上下文。自动断开所有 MCP 服务器的连接。

参数:

exc_type -- 发生异常时的异常类型（否则为 None）
exc_val -- 发生异常时的异常实例（否则为 None）
exc_tb -- 发生异常时的回溯（否则为 None）

class evofabric.core.tool.McpSessionController(BaseComponent)[源代码]#

管理MCP服务器连接生命周期和会话操作的控制器类。负责建立连接、管理会话状态以及使用异步上下文管理器实现优雅断开连接。

参数:

server_link (McpServerLink) -- MCP服务器通信链路对象
server_name (str) -- 连接的MCP服务器的名称标识符
persistent_link (bool) -- 设为True时，退出上下文（aexit）不会断开MCP服务器连接。默认为False

property session#: 返回当前活动的客户端会话（只读属性）。

property is_connect#: 返回连接状态（True/False），用于监控目的。

async connect() → None[源代码]#: 启动并等待MCP服务器连接建立。创建后台任务维护会话循环，阻塞直至连接完全建立。安全地支持多次调用，重复调用会提前返回。

async disconnect() → None[源代码]#: 终止MCP连接。取消会话任务，等待清理完成，并重置所有连接状态。

async _run_session_loop(ready_event: asyncio.Event) → None[源代码]#

内部方法：运行会话维护的主循环。建立MCP会话后等待连接就绪事件，进入持久等待状态直至被取消。

参数:: ready_event -- 用于通知连接已就绪的Event对象

async __aenter__() → McpSessionController[源代码]#: 异步上下文管理器入口。自动调用connect()方法建立连接，返回自身实例。

async __aexit__(exc_type: Any, exc_val: Any, exc_tb: Any) → None[源代码]#: 异步上下文管理器出口。仅当persistent_link为False时调用disconnect()方法断开连接。

Code Sandbox#

class evofabric.core.tool.CodeSandbox(BaseComponent)[源代码]#

安全的 Python 代码沙箱，基于 Docker 实现，用于在隔离环境中执行 Python 代码。

继承自 BaseComponent。

参数:: config (CodeExecDockerConfig) -- 沙箱配置，包括镜像、容器名称、工作目录等信息

使用示例：

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()[源代码]#: 根据配置创建并启动 Docker 容器。

run_python(code: str)[源代码]#

在沙箱容器中执行 Python 代码。

参数:: code (str) -- 待执行的 Python 代码
返回:: 执行结果
返回类型:: ExecResult

run_cmd(cmd: str)[源代码]#

在沙箱容器中执行 shell 命令。

参数:: cmd (str) -- 待执行的 shell 命令
返回:: 执行结果
返回类型:: ExecResult

stop()[源代码]#: 停止 Docker 容器。如果配置中的 auto_remove 为 True，容器停止后将被自动移除。

Tool Controller#

class evofabric.core.tool.ToolController(BaseComponent)[源代码]#

工具控制器，用于管理工具的激活和禁用状态。继承自 :py:class`~evofabric.core.factory.BaseComponent`

参数:

default_mode (Literal['activate', 'deactivate']) -- 默认工具行为，当没有规则匹配时使用
rules (list[Union[ToolControlPattern, dict]]) -- 用于控制工具激活/禁用的规则列表，按顺序应用规则，第一个匹配规则决定工具状态

类属性默认值说明：

default_mode 的默认值为 "activate"。可选值：

"activate": 自动激活没有匹配规则的工具（默认值）
"deactivate": 自动禁用没有匹配规则的工具

rules 的默认值为空字典。每个规则可以是：

带有以下属性的 ToolControlPattern 实例：

mode: 'activate' 或 'deactivate'

pattern: glob 通配符字符串

带有以下键的字典：

'mode': 'activate' 或 'deactivate'

'pattern': glob 通配符字符串

备注

使用 McpToolManager 时的注意事项：

实际工具名称格式为 [server_name]_[tool_name]
通配符模式必须以 server_name 作为前缀
示例：为匹配来自 "math" 服务器的所有工具，使用模式 "math_*" （匹配 "math_calculator"、"math_grapher" 等）

模式示例：

math_* → 匹配来自 "math" 服务器的所有工具（例如 "math_calculator"、"math_grapher"）
text_* → 匹配来自 "text" 服务器的所有工具
math_calculator → 仅匹配此特定工具
*_calculator → 匹配任何以 "*_calculator" 结尾的工具（来自任何服务器）

check_tool_status(tool_name: str) → bool[源代码]#

根据应用规则检查工具是否处于激活状态。

参数:: tool_name (str) -- 要检查的工具名称
返回:: 如果工具激活返回 True，否则返回 False
返回类型:: bool

filter_tool_list(tool_list: List[Dict]) → List[Dict][源代码]#

过滤工具列表，仅包含激活的工具。

参数:: tool_list (List[Dict]) -- 工具字典列表
返回:: 激活工具的列表
返回类型:: List[Dict]

activate_tool(tool_name: str)[源代码]#

激活特定工具，赋予最高优先级。移除此工具名称的任何现有规则。

参数:: tool_name (str) -- 要激活的工具的确切名称

deactivate_tool(tool_name: str)[源代码]#

禁用特定工具，赋予最高优先级。移除此工具名称的任何现有规则。

参数:: tool_name (str) -- 要禁用的工具的确切名称

Utils#

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)[源代码]#

将Python可调用对象转换为LLM工具使用的JSON模式。此函数解析函数签名、文档字符串和参数默认值，创建与LLM工具调用格式兼容的完整接口描述。

支持的函数类型包括：普通函数、类方法、@classmethod、@staticmethod、函数/类方法的partial对象、lambda函数。

参数:

function (Callable) -- 要转换为模式的可调用函数
name (Optional[str]) -- 自定义函数名，未提供则使用函数的`__name__`属性
description (Optional[str]) -- 自定义描述，未提供则使用文档字符串摘要
include_long_description (bool) -- 是否包含文档字符串中的长描述
include_var_positional (bool) -- 是否包含位置可变参数(*args)
include_var_keyword (bool) -- 是否包含关键字可变参数(**kwargs)
exclude_params (Optional[List[str]]) -- 需要排除的参数名称列表

返回:

包含OpenAI兼容函数模式的元组

返回类型:

tuple

Return1:

包含完整函数模式的字典

Rtype1:

dict

Return2:

实际被排除的参数名称列表

Rtype2:

List[str]

evofabric.core.tool

目录

evofabric.core.tool#

Base Tool#

Tool Manager#

Code Sandbox#

Tool Controller#

Utils#