`jianmu.tool`¶

适用对象：工具作者 / Agent 应用开发者 / 执行层维护者是否必读：是相关模块：jianmu.execution, jianmu.guard, jianmu.node

1. 模块职责¶

jianmu.tool 是工具系统的公开入口，负责暴露工具基类、函数装饰器、工具集合门面和常用内置工具。

这是“声明工具”和“把工具接到 Agent/节点里”的主入口。

2. 适合查什么¶

核心对象：Tool、ToolCall、ToolResult
工具节点：jianmu.node.ToolNode
工具集合：ToolSet
装饰器：tool、FunctionTool
高级扩展：ToolProvider
内置工具：CalculatorTool、PythonREPLTool、FileReadTool

3. 注意事项¶

HTTPTool、DuckDuckGoSearchTool 依赖可选包
声明工具时优先依赖 Tool / tool / ToolSet，不要直接耦合内部 runner
真正的执行隔离、审批和沙箱逻辑主要属于 jianmu.execution 与 jianmu.guard

4. builtin 路径说明¶

jianmu.tool.builtin 表示“这些工具当前由框架内建提供”，它更像源码组织层，而不是推荐用户长期依赖的主导入层。

因此新代码应优先写成：

from jianmu.tool import CalculatorTool
from jianmu.tool import PythonREPLTool
from jianmu.tool import FileReadTool

而不是默认写成：

from jianmu.tool.builtin import CalculatorTool
from jianmu.tool.builtin import PythonREPLTool

如果未来内建工具的文件布局调整，jianmu.tool 这一层更容易保持稳定。

5. 最小示例¶

from jianmu.tool import tool


@tool
def echo(input: str) -> str:
    """Return the input unchanged."""
    return input


tools = [echo]

from jianmu.tool import ToolSet

toolset = ToolSet.from_tools([echo])

6. 常见入口¶

想快速把函数暴露成工具：看 tool / FunctionTool
想写类式工具：看 Tool
想组合和执行一组工具：看 ToolSet
想接入动态工具来源：看 ToolProvider
想执行 agent 发出的工具调用：看 ToolExecutor

7. API 参考¶

tool ¶

jianmu tools module - Tool base class and built-in tools.

ParallelDecision `dataclass` ¶

ParallelDecision(
    allowed: bool, reason: str, category: str = ""
)

Explain whether one concrete tool call may run concurrently.

Tool ¶

Bases: ABC

Base class for callable tools exposed to agents and workflow nodes.

Subclass Tool when you want a reusable, named capability that can be:

called directly by ToolExecutor
exposed to an LLM as function-calling schema
wrapped into a workflow node via as_node()

Minimal example::

class EchoTool(Tool):
    name = "echo"
    description = "Return the input text unchanged."

    def run(self, input: str) -> str:
        return input

属性：

名称	类型	描述
`name`	`str`	Public tool name exposed to agents and executors.
`description`	`str`	Human-readable tool description.
`input_schema`	`Dict[str, Any]`	JSON schema describing accepted tool inputs.
`output_schema`	`Dict[str, Any]`	JSON schema describing tool outputs.
`parallel_safe`	`bool`	Whether the tool may execute concurrently with peers.
`effect_tags`	`tuple[str, ...]`	Normalized side-effect tags used by orchestration logic.

run `abstractmethod` ¶

run(input: Any) -> Any

Execute the tool with given input.

参数：

名称	类型	描述	默认
`input`	`Any`	Parsed tool input. For function-style tools this may be a structured dictionary rather than a raw string.	必需

返回：

类型	描述
`Any`	Tool result to be converted into an observation.

源代码位于： jianmu/tool/base.py

@abstractmethod
def run(self, input: Any) -> Any:
    """Execute the tool with given input.

    Args:
        input: Parsed tool input. For function-style tools this may be a
            structured dictionary rather than a raw string.

    Returns:
        Tool result to be converted into an observation.
    """
    pass

execute `async` ¶

execute(
    args: Any,
    *,
    injected: Optional[Dict[str, Any]] = None,
    prefer_injected: bool = True,
    runner: Optional[Any] = None,
    runner_event_context: Optional[Any] = None,
) -> Any

Execute this tool through the shared Jianmu execution contract.

参数：

名称	类型	描述	默认
`args`	`Any`	Explicit tool arguments.	必需
`injected`	`Optional[Dict[str, Any]]`	Optional runtime-injected arguments merged into `args`.	`None`
`prefer_injected`	`bool`	Whether injected values override explicit args.	`True`
`runner`	`Optional[Any]`	Optional sandbox-aware runner delegated to perform execution.	`None`
`runner_event_context`	`Optional[Any]`	Optional runtime-event correlation metadata forwarded to the shared runner when present.	`None`

返回：

类型	描述
`Any`	Tool result produced by the shared execution contract.

源代码位于： jianmu/tool/base.py

async def execute(
    self,
    args: Any,
    *,
    injected: Optional[Dict[str, Any]] = None,
    prefer_injected: bool = True,
    runner: Optional[Any] = None,
    runner_event_context: Optional[Any] = None,
) -> Any:
    """Execute this tool through the shared Jianmu execution contract.

    Args:
        args: Explicit tool arguments.
        injected: Optional runtime-injected arguments merged into ``args``.
        prefer_injected: Whether injected values override explicit args.
        runner: Optional sandbox-aware runner delegated to perform execution.
        runner_event_context: Optional runtime-event correlation metadata
            forwarded to the shared runner when present.

    Returns:
        Tool result produced by the shared execution contract.
    """
    call_args = self._merge_call_args(args, injected=injected, prefer_injected=prefer_injected)
    if runner is not None:
        return await runner.run(self, call_args, event_context=runner_event_context)
    return await self._execute_local(call_args)

permission_check_for_call ¶

permission_check_for_call(
    args: Any, *, context: ToolPermissionContext
) -> PermissionCheck | None

Project one concrete invocation into Permission V2.

Custom tools may override this side-effect-free method. The default implementation uses the declared permission_action and falls back to the tool name plus an exact canonical-arguments resource.

源代码位于： jianmu/tool/base.py

def permission_check_for_call(
    self,
    args: Any,
    *,
    context: "ToolPermissionContext",
) -> "PermissionCheck | None":
    """Project one concrete invocation into Permission V2.

    Custom tools may override this side-effect-free method. The default
    implementation uses the declared ``permission_action`` and falls back
    to the tool name plus an exact canonical-arguments resource.
    """
    from jianmu.guard.permission.tool_projection import project_tool_permission

    return project_tool_permission(self, args, context=context)

parallel_decision_for_call ¶

parallel_decision_for_call(
    args: Any, *, runtime_injected: bool
) -> ParallelDecision

Return the default concurrency decision for one concrete call.

Subclasses may override this method when concurrency safety depends on the call arguments. The default deliberately preserves Jianmu's previous conservative behavior for runtime-injected and stateful tools. Overrides must be deterministic and side-effect free because planners and observability code may query the decision more than once.

源代码位于： jianmu/tool/base.py

def parallel_decision_for_call(
    self,
    args: Any,
    *,
    runtime_injected: bool,
) -> ParallelDecision:
    """Return the default concurrency decision for one concrete call.

    Subclasses may override this method when concurrency safety depends on
    the call arguments. The default deliberately preserves Jianmu's
    previous conservative behavior for runtime-injected and stateful tools.
    Overrides must be deterministic and side-effect free because planners
    and observability code may query the decision more than once.
    """
    del args
    if runtime_injected:
        return ParallelDecision(
            allowed=False,
            reason="tool requires runtime-injected state",
            category="runtime_injected",
        )
    if not bool(self.parallel_safe):
        return ParallelDecision(
            allowed=False,
            reason="tool is not marked parallel-safe",
            category="stateful_tool",
        )
    return ParallelDecision(
        allowed=True,
        reason="tool call is safe for concurrent execution",
        category="allowed",
    )

as_node ¶

as_node(
    name: Optional[str] = None,
    *,
    input_key: str = "input",
    output_key: str = "output",
    execute: Optional[bool] = None,
    namespace: str | None = None,
) -> Any

Wrap this tool into a ToolNode for use in a behavior tree.

参数：

名称	类型	描述	默认
`name`	`Optional[str]`	Node name override; defaults to `self.name`.	`None`
`input_key`	`str`	State key from which the tool reads its input.	`'input'`
`output_key`	`str`	State key to which the tool writes its output.	`'output'`
`execute`	`Optional[bool]`	Whether the node should run the tool when ticked.	`None`
`namespace`	`str \| None`	Optional state namespace for port resolution.	`None`

返回：

类型	描述
`Any`	A `ToolNode` instance wrapping this tool.

This is useful when the same tool should be available both to LLM agent loops and to explicit workflow branches.

源代码位于： jianmu/tool/base.py

def as_node(
    self,
    name: Optional[str] = None,
    *,
    input_key: str = "input",
    output_key: str = "output",
    execute: Optional[bool] = None,
    namespace: str | None = None,
) -> Any:
    """Wrap this tool into a ``ToolNode`` for use in a behavior tree.

    Args:
        name: Node name override; defaults to ``self.name``.
        input_key: State key from which the tool reads its input.
        output_key: State key to which the tool writes its output.
        execute: Whether the node should run the tool when ticked.
        namespace: Optional state namespace for port resolution.

    Returns:
        A ``ToolNode`` instance wrapping this tool.

    This is useful when the same tool should be available both to LLM agent
    loops and to explicit workflow branches.
    """
    from jianmu.node.builtin.tool import ToolNode
    return ToolNode(
        name=name or self.name,
        tool=self,
        input_key=input_key,
        output_key=output_key,
        execute=execute,
        namespace=namespace,
    )

get_description ¶

get_description() -> str

Resolve the best available description for prompts and schemas.

返回：

类型	描述
`str`	The class-level `description` attribute when non-default, otherwise
`str`	the cleaned class docstring, or an empty string.

源代码位于： jianmu/tool/base.py

def get_description(self) -> str:
    """Resolve the best available description for prompts and schemas.

    Returns:
        The class-level ``description`` attribute when non-default, otherwise
        the cleaned class docstring, or an empty string.
    """
    if self.description and self.description != "No description provided":
        return self.description
    doc = self._docstring_description()
    if doc:
        return doc
    return ""

get_effect_tags ¶

get_effect_tags() -> tuple[str, ...]

Return normalized effect tags for downstream orchestration logic.

返回：

类型	描述
`tuple[str, ...]`	Normalized effect-tag tuple with empty values removed.

源代码位于： jianmu/tool/base.py

def get_effect_tags(self) -> tuple[str, ...]:
    """Return normalized effect tags for downstream orchestration logic.

    Returns:
        Normalized effect-tag tuple with empty values removed.
    """
    raw_tags = self.effect_tags or ()
    if isinstance(raw_tags, str):
        raw_tags = (raw_tags,)
    normalized: list[str] = []
    for tag in raw_tags:
        value = str(tag or "").strip().lower()
        if value:
            normalized.append(value)
    return tuple(normalized)

to_schema ¶

to_schema() -> Dict[str, Any]

Return the JSON function schema (name/description/parameters).

返回：

类型	描述
`Dict[str, Any]`	Dictionary with `name`, `description`, and `parameters`
`Dict[str, Any]`	suitable for OpenAI-style function-calling.

源代码位于： jianmu/tool/base.py

def to_schema(self) -> Dict[str, Any]:
    """Return the JSON function schema (name/description/parameters).

    Returns:
        Dictionary with ``name``, ``description``, and ``parameters``
        suitable for OpenAI-style function-calling.
    """
    description = self.get_description()
    if not description:
        description = self.name or "tool"
    return {
        "name": self.name,
        "description": description,
        "parameters": self._normalize_parameters(),
    }

spec ¶

spec() -> Dict[str, Any]

Return a normalized tool spec for prompts and UIs.

返回：

类型	描述
`Dict[str, Any]`	Dictionary with `name`, `description`, `input_schema`,
`Dict[str, Any]`	`output_schema`, `parameters`, and `returns`.

源代码位于： jianmu/tool/base.py

def spec(self) -> Dict[str, Any]:
    """Return a normalized tool spec for prompts and UIs.

    Returns:
        Dictionary with ``name``, ``description``, ``input_schema``,
        ``output_schema``, ``parameters``, and ``returns``.
    """
    return {
        "name": self.name,
        "description": self.get_description(),
        "input_schema": self.input_schema,
        "output_schema": self.output_schema,
        "parameters": self._normalize_parameters(),
        "returns": self._normalize_output_schema(),
    }

ToolSet `dataclass` ¶

ToolSet(tools: list[Tool] = list())

Facade over a named collection of tools.

ToolSet is the main public entrypoint for assembling tools from:

builtin tool names
explicit tool instances
dynamic ToolProvider sources

属性：

名称	类型	描述
`tools`	`list[Tool]`	Tool instances in execution and prompt order.

from_tools `classmethod` ¶

from_tools(
    tools: Iterable[Tool] | None = None,
) -> "ToolSet"

Build a tool set from explicit tool instances.

参数：

名称	类型	描述	默认
`tools`	`Iterable[Tool] \| None`	Tool instances to include.	`None`

返回：

类型	描述
`'ToolSet'`	Tool set containing the provided explicit tool instances.

源代码位于： jianmu/tool/set.py

@classmethod
def from_tools(cls, tools: Iterable[Tool] | None = None) -> "ToolSet":
    """Build a tool set from explicit tool instances.

    Args:
        tools: Tool instances to include.

    Returns:
        Tool set containing the provided explicit tool instances.
    """
    return cls(list(cls._merge_groups(list(tools or []))))

resolve `classmethod` ¶

resolve(
    names: list[str],
    *,
    custom_tools: list[Tool] | None = None,
) -> "ToolSet"

Resolve builtin/custom tool names into a concrete tool set.

参数：

名称	类型	描述	默认
`names`	`list[str]`	Builtin or custom tool names to resolve.	必需
`custom_tools`	`list[Tool] \| None`	Additional tool instances to consider.	`None`

返回：

类型	描述
`'ToolSet'`	Tool set containing resolved builtin and custom tools.

源代码位于： jianmu/tool/set.py

@classmethod
def resolve(
    cls,
    names: list[str],
    *,
    custom_tools: list[Tool] | None = None,
) -> "ToolSet":
    """Resolve builtin/custom tool names into a concrete tool set.

    Args:
        names: Builtin or custom tool names to resolve.
        custom_tools: Additional tool instances to consider.

    Returns:
        Tool set containing resolved builtin and custom tools.
    """
    return cls(BuiltinToolProvider.resolve_named_tools(names, custom_tools=custom_tools))

from_providers `classmethod` ¶

from_providers(
    providers: Iterable[ToolProvider], **kwargs: Any
) -> "ToolSet"

Collect tools from dynamic providers and wrap them as a tool set.

参数：

名称	类型	描述	默认
`providers`	`Iterable[ToolProvider]`	Tool providers to include.	必需
`**kwargs`	`Any`	Provider-specific resolution arguments.	`{}`

返回：

类型	描述
`'ToolSet'`	Tool set containing the merged provider tools.

源代码位于： jianmu/tool/set.py

@classmethod
def from_providers(
    cls,
    providers: Iterable[ToolProvider],
    **kwargs: Any,
) -> "ToolSet":
    """Collect tools from dynamic providers and wrap them as a tool set.

    Args:
        providers: Tool providers to include.
        **kwargs: Provider-specific resolution arguments.

    Returns:
        Tool set containing the merged provider tools.
    """
    return cls(cls._collect_provider_tools(providers, **kwargs))

list_builtin_names `classmethod` ¶

list_builtin_names() -> list[str]

List builtin tool names visible to the registry.

返回：

类型	描述
`list[str]`	Canonical builtin tool names visible to the registry.

源代码位于： jianmu/tool/set.py

@classmethod
def list_builtin_names(cls) -> list[str]:
    """List builtin tool names visible to the registry.

    Returns:
        Canonical builtin tool names visible to the registry.
    """
    return BuiltinToolProvider.list_builtin_names()

merge ¶

merge(*groups: Iterable[Tool] | 'ToolSet') -> 'ToolSet'

Return a new tool set merged with explicit tools or other sets.

参数：

名称	类型	描述	默认
`*groups`	`Iterable[Tool] \| 'ToolSet'`	Additional tool groups or tool sets to merge.	`()`

返回：

类型	描述
`'ToolSet'`	New tool set containing the merged tool collection.

源代码位于： jianmu/tool/set.py

def merge(self, *groups: Iterable[Tool] | "ToolSet") -> "ToolSet":
    """Return a new tool set merged with explicit tools or other sets.

    Args:
        *groups: Additional tool groups or tool sets to merge.

    Returns:
        New tool set containing the merged tool collection.
    """
    merged_groups: list[Iterable[Tool]] = [self.tools]
    for group in groups:
        if isinstance(group, ToolSet):
            merged_groups.append(group.tools)
        else:
            merged_groups.append(group)
    return ToolSet(list(self._merge_groups(*merged_groups)))

with_tools ¶

with_tools(tools: Iterable[Tool]) -> 'ToolSet'

Return a new tool set with more explicit tools merged in.

参数：

名称	类型	描述	默认
`tools`	`Iterable[Tool]`	Tool instances to include.	必需

返回：

类型	描述
`'ToolSet'`	New tool set with the provided explicit tools merged in.

源代码位于： jianmu/tool/set.py

def with_tools(self, tools: Iterable[Tool]) -> "ToolSet":
    """Return a new tool set with more explicit tools merged in.

    Args:
        tools: Tool instances to include.

    Returns:
        New tool set with the provided explicit tools merged in.
    """
    return self.merge(list(tools))

get ¶

get(name: str) -> Tool | None

Return one tool by name, or None when absent.

参数：

名称	类型	描述	默认
`name`	`str`	Tool name to resolve.	必需

返回：

类型	描述
`Tool \| None`	Matching tool instance, or `None` when absent.

源代码位于： jianmu/tool/set.py

def get(self, name: str) -> Tool | None:
    """Return one tool by name, or ``None`` when absent.

    Args:
        name: Tool name to resolve.

    Returns:
        Matching tool instance, or ``None`` when absent.
    """
    target = str(name).strip().lower()
    for tool in self.tools:
        tool_name = str(getattr(tool, "name", "") or "").strip().lower()
        if tool_name == target:
            return tool
    return None

execute `async` ¶

execute(
    name: str,
    args: Any,
    *,
    injected: dict[str, Any] | None = None,
    prefer_injected: bool = True,
    runner: Any | None = None,
) -> Any

Execute one named tool from this set.

参数：

名称	类型	描述	默认
`name`	`str`	Tool name to execute.	必需
`args`	`Any`	Explicit tool arguments.	必需
`injected`	`dict[str, Any] \| None`	Optional runtime-injected arguments.	`None`
`prefer_injected`	`bool`	Whether injected values override explicit values.	`True`
`runner`	`Any \| None`	Optional delegated execution runner.	`None`

返回：

类型	描述
`Any`	Tool result produced by the resolved tool.

引发：

类型	描述
`KeyError`	If a required key is missing.

源代码位于： jianmu/tool/set.py

async def execute(
    self,
    name: str,
    args: Any,
    *,
    injected: dict[str, Any] | None = None,
    prefer_injected: bool = True,
    runner: Any | None = None,
) -> Any:
    """Execute one named tool from this set.

    Args:
        name: Tool name to execute.
        args: Explicit tool arguments.
        injected: Optional runtime-injected arguments.
        prefer_injected: Whether injected values override explicit values.
        runner: Optional delegated execution runner.

    Returns:
        Tool result produced by the resolved tool.

    Raises:
        KeyError: If a required key is missing.
    """
    tool = self.get(name)
    if tool is None:
        raise KeyError(f"Tool not found: {name}")
    return await tool.execute(
        args,
        injected=injected,
        prefer_injected=prefer_injected,
        runner=runner,
    )

is_parallel_safe ¶

is_parallel_safe(names: Iterable[str]) -> bool

Return whether all named tools may run in parallel safely.

参数：

名称	类型	描述	默认
`names`	`Iterable[str]`	Tool names to validate for parallel execution.	必需

返回：

类型	描述
`bool`	`True` if every named tool exists and is parallel-safe.

源代码位于： jianmu/tool/set.py

def is_parallel_safe(self, names: Iterable[str]) -> bool:
    """Return whether all named tools may run in parallel safely.

    Args:
        names: Tool names to validate for parallel execution.

    Returns:
        ``True`` if every named tool exists and is parallel-safe.
    """
    for name in names:
        tool = self.get(str(name))
        if tool is None:
            return False
        if not bool(getattr(tool, "parallel_safe", True)):
            return False
    return True

schemas ¶

schemas() -> list[dict[str, Any]]

Return normalized function-calling schemas for all tools.

返回：

类型	描述
`list[dict[str, Any]]`	Normalized function-calling schemas for all tools in the set.

源代码位于： jianmu/tool/set.py

def schemas(self) -> list[dict[str, Any]]:
    """Return normalized function-calling schemas for all tools.

    Returns:
        Normalized function-calling schemas for all tools in the set.
    """
    return [tool.to_schema() for tool in self.tools]

specs ¶

specs() -> list[dict[str, Any]]

Return normalized tool specs for prompts and UIs.

返回：

类型	描述
`list[dict[str, Any]]`	Normalized tool specs for all tools in the set.

源代码位于： jianmu/tool/set.py

def specs(self) -> list[dict[str, Any]]:
    """Return normalized tool specs for prompts and UIs.

    Returns:
        Normalized tool specs for all tools in the set.
    """
    return [tool.spec() for tool in self.tools]

describe ¶

describe() -> str

Render a human-readable bullet list for prompt construction.

返回：

类型	描述
`str`	Human-readable bullet list describing the tools.

源代码位于： jianmu/tool/set.py

def describe(self) -> str:
    """Render a human-readable bullet list for prompt construction.

    Returns:
        Human-readable bullet list describing the tools.
    """
    if not self.tools:
        return ""
    return "\n".join(
        f"- {tool.name}: {tool.get_description()}"
        for tool in self.tools
        if getattr(tool, "name", None)
    )

names ¶

names() -> list[str]

Return tool names in prompt/execution order.

返回：

类型	描述
`list[str]`	Tool names in execution and prompt order.

源代码位于： jianmu/tool/set.py

def names(self) -> list[str]:
    """Return tool names in prompt/execution order.

    Returns:
        Tool names in execution and prompt order.
    """
    return [tool.name for tool in self.tools if getattr(tool, "name", None)]

BuiltinToolProvider ¶

Bases: BaseToolProvider

Builtin Jianmu tool registry and lazy resolution provider.

属性：

名称	类型	描述
`_builtins`	`Dict[str, Type[Tool]]`	Canonical builtin tool registry keyed by tool name.
`_aliases`	`Dict[str, str]`	Alias mapping from alternate names to canonical builtin names.
`_builtins_loaded`		Whether lazy builtin registration has completed.

aliases `classmethod` ¶

aliases() -> Dict[str, str]

Return the current alias-to-canonical tool name mapping.

返回：

类型	描述
`Dict[str, str]`	The resulting `Dict[str, str]` value.

源代码位于： jianmu/tool/provider.py

@classmethod
def aliases(cls) -> Dict[str, str]:
    """Return the current alias-to-canonical tool name mapping.

    Returns:
        The resulting `Dict[str, str]` value.
    """
    cls._ensure_builtins()
    return cls._aliases

register_tool `classmethod` ¶

register_tool(name: str, tool_cls: Type[Tool]) -> None

Register a builtin tool class under its canonical name.

参数：

名称	类型	描述	默认
`name`	`str`	Canonical builtin tool name.	必需
`tool_cls`	`Type[Tool]`	Tool class registered under the canonical name.	必需

源代码位于： jianmu/tool/provider.py

@classmethod
def register_tool(cls, name: str, tool_cls: Type[Tool]) -> None:
    """Register a builtin tool class under its canonical name.

    Args:
        name: Canonical builtin tool name.
        tool_cls: Tool class registered under the canonical name.
    """
    cls._builtins[str(name)] = tool_cls

register_alias `classmethod` ¶

register_alias(alias: str, canonical: str) -> None

Register an alias that resolves to a canonical builtin tool name.

参数：

名称	类型	描述	默认
`alias`	`str`	Alternate name that should resolve to a builtin tool.	必需
`canonical`	`str`	Canonical builtin tool name referenced by the alias.	必需

源代码位于： jianmu/tool/provider.py

@classmethod
def register_alias(cls, alias: str, canonical: str) -> None:
    """Register an alias that resolves to a canonical builtin tool name.

    Args:
        alias: Alternate name that should resolve to a builtin tool.
        canonical: Canonical builtin tool name referenced by the alias.
    """
    cls._aliases[str(alias)] = str(canonical)

canonical_tool_name `classmethod` ¶

canonical_tool_name(name: str) -> str

Resolve an input name through the alias table when available.

参数：

名称	类型	描述	默认
`name`	`str`	Requested tool name or alias.	必需

返回：

类型	描述
`str`	Canonical builtin tool name when an alias exists, otherwise `name`.

源代码位于： jianmu/tool/provider.py

@classmethod
def canonical_tool_name(cls, name: str) -> str:
    """Resolve an input name through the alias table when available.

    Args:
        name: Requested tool name or alias.

    Returns:
        Canonical builtin tool name when an alias exists, otherwise ``name``.
    """
    cls._ensure_builtins()
    return cls._aliases.get(name, name)

list_builtin_names `classmethod` ¶

list_builtin_names() -> List[str]

Return the canonical names of all registered builtin tools.

返回：

类型	描述
`List[str]`	Canonical names of all registered builtin tools.

源代码位于： jianmu/tool/provider.py

@classmethod
def list_builtin_names(cls) -> List[str]:
    """Return the canonical names of all registered builtin tools.

    Returns:
        Canonical names of all registered builtin tools.
    """
    cls._ensure_builtins()
    return list(cls._builtins.keys())

is_builtin_tool_name `classmethod` ¶

is_builtin_tool_name(name: str) -> bool

Return whether the provided name resolves to a builtin tool.

参数：

名称	类型	描述	默认
`name`	`str`	Requested tool name or alias.	必需

返回：

类型	描述
`bool`	`True` if the name resolves to a builtin tool; otherwise `False`.

源代码位于： jianmu/tool/provider.py

@classmethod
def is_builtin_tool_name(cls, name: str) -> bool:
    """Return whether the provided name resolves to a builtin tool.

    Args:
        name: Requested tool name or alias.

    Returns:
        ``True`` if the name resolves to a builtin tool; otherwise ``False``.
    """
    cls._ensure_builtins()
    canonical = cls.canonical_tool_name(name)
    return canonical in cls._builtins

resolve_named_tools `classmethod` ¶

resolve_named_tools(
    tool_names: List[str],
    *,
    custom_tools: List[Tool] | None = None,
) -> List[Tool]

Resolve named builtin and custom tools into instantiated tool objects.

参数：

名称	类型	描述	默认
`tool_names`	`List[str]`	Tool names to resolve.	必需
`custom_tools`	`List[Tool] \| None`	Additional tool instances to consider.	`None`

返回：

类型	描述
`List[Tool]`	Instantiated tool objects with deterministic name-based deduplication.

源代码位于： jianmu/tool/provider.py

@classmethod
def resolve_named_tools(
    cls,
    tool_names: List[str],
    *,
    custom_tools: List[Tool] | None = None,
) -> List[Tool]:
    """Resolve named builtin and custom tools into instantiated tool objects.

    Args:
        tool_names: Tool names to resolve.
        custom_tools: Additional tool instances to consider.

    Returns:
        Instantiated tool objects with deterministic name-based deduplication.
    """
    cls._ensure_builtins()

    custom_map: Dict[str, Tool] = {}
    for tool in custom_tools or []:
        custom_map[tool.name.lower()] = tool

    seen: Dict[str, Tool] = {}
    for name in tool_names:
        canonical = cls.canonical_tool_name(name)
        key = canonical.lower()
        if key in seen:
            continue

        if key in custom_map:
            seen[key] = custom_map[key]
            continue

        tool_cls = cls._builtins.get(canonical)
        if tool_cls:
            try:
                seen[key] = tool_cls()
            except Exception:
                continue

    for key, tool in custom_map.items():
        if key not in seen:
            seen[key] = tool

    return list(seen.values())

get_tools `classmethod` ¶

get_tools(**kwargs: Any) -> list[Tool]

Return builtin tools, optionally narrowed to the requested names.

参数：

名称	类型	描述	默认
`**kwargs`	`Any`	Optional `names` and `custom_tools` resolution inputs.	`{}`

返回：

类型	描述
`list[Tool]`	Builtin tools, optionally narrowed to the requested names.

源代码位于： jianmu/tool/provider.py

@classmethod
def get_tools(cls, **kwargs: Any) -> list[Tool]:
    """Return builtin tools, optionally narrowed to the requested names.

    Args:
        **kwargs: Optional ``names`` and ``custom_tools`` resolution inputs.

    Returns:
        Builtin tools, optionally narrowed to the requested names.
    """
    names = kwargs.get("names")
    custom_tools = kwargs.get("custom_tools")
    if names is None:
        return cls.resolve_named_tools(cls.list_builtin_names(), custom_tools=custom_tools)
    return cls.resolve_named_tools(list(names), custom_tools=custom_tools)

ToolProvider ¶

Bases: Protocol

Lifecycle and resolution contract for dynamic tool sources.

Implement a provider when the available tool set depends on runtime context, external connections, or lazy initialization. Static tool lists do not need a provider; pass them directly to presets or executors instead.

initialize `async` ¶

initialize() -> None

Initialize provider resources before tool resolution.

源代码位于： jianmu/tool/provider.py

async def initialize(self) -> None:
    """Initialize provider resources before tool resolution."""
    ...

get_tools ¶

get_tools(**kwargs: Any) -> list[Tool]

Return tools currently exposed by the provider.

参数：

名称	类型	描述	默认
`**kwargs`	`Any`	Provider-specific resolution arguments.	`{}`

返回：

类型	描述
`list[Tool]`	List of Tool instances.

源代码位于： jianmu/tool/provider.py

def get_tools(self, **kwargs: Any) -> list[Tool]:
    """Return tools currently exposed by the provider.

    Args:
        **kwargs: Provider-specific resolution arguments.

    Returns:
        List of Tool instances.
    """
    ...

close `async` ¶

close() -> None

Release provider resources.

源代码位于： jianmu/tool/provider.py

async def close(self) -> None:
    """Release provider resources."""
    ...

BaseToolProvider ¶

No-op lifecycle base class shared by single/multi-agent tool providers.

initialize `async` ¶

initialize() -> None

Perform no-op initialization.

源代码位于： jianmu/tool/provider.py

async def initialize(self) -> None:
    """Perform no-op initialization."""
    return

close `async` ¶

close() -> None

Perform no-op cleanup.

源代码位于： jianmu/tool/provider.py

async def close(self) -> None:
    """Perform no-op cleanup."""
    return

SuspendedToolCall `dataclass` ¶

SuspendedToolCall(
    tool: str,
    arguments: Any = None,
    tool_call_id: str = "",
    request_id: str = "",
)

Persisted description of one suspended tool call.

属性：

名称	类型	描述
`tool`	`str`	Canonical tool name associated with the suspension.
`arguments`	`Any`	Serialized arguments for the suspended tool call.
`tool_call_id`	`str`	Stable tool-call identifier when available.
`request_id`	`str`	Suspension request identifier associated with the tool call.

from_dict `classmethod` ¶

from_dict(
    data: dict[str, Any] | None,
) -> "SuspendedToolCall | None"

Parse one persisted suspended tool-call record.

参数：

名称	类型	描述	默认
`data`	`dict[str, Any] \| None`	Serialized suspended tool-call payload.	必需

返回：

类型	描述
`'SuspendedToolCall \| None'`	Parsed suspended tool call, or `None` when the payload is invalid.

源代码位于： jianmu/tool/resume.py

@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> "SuspendedToolCall | None":
    """Parse one persisted suspended tool-call record.

    Args:
        data: Serialized suspended tool-call payload.

    Returns:
        Parsed suspended tool call, or ``None`` when the payload is invalid.
    """
    if not isinstance(data, dict):
        return None
    tool = str(data.get("tool") or "").strip().lower()
    if not tool:
        return None
    return cls(
        tool=tool,
        arguments=copy.deepcopy(data.get("arguments")),
        tool_call_id=str(data.get("id") or data.get("tool_call_id") or "").strip(),
        request_id=str(data.get("request_id") or "").strip(),
    )

to_dict ¶

to_dict() -> dict[str, Any]

Serialize one suspended tool-call record.

返回：

类型	描述
`dict[str, Any]`	Dictionary payload suitable for runtime metadata storage.

源代码位于： jianmu/tool/resume.py

def to_dict(self) -> dict[str, Any]:
    """Serialize one suspended tool-call record.

    Returns:
        Dictionary payload suitable for runtime metadata storage.
    """
    return {
        "tool": self.tool,
        "arguments": copy.deepcopy(self.arguments),
        "id": self.tool_call_id,
        "request_id": self.request_id,
    }

SuspendedToolCallStore ¶

SuspendedToolCallStore(interaction: Any)

Interaction-bound store for suspended tool-call metadata.

属性：

名称	类型	描述
`_interaction`		Interaction object used to read and write runtime metadata.

Bind the store to one interaction instance.

参数：

名称	类型	描述	默认
`interaction`	`Any`	Interaction object exposing runtime metadata helpers.	必需

源代码位于： jianmu/tool/resume.py

def __init__(self, interaction: Any) -> None:
    """Bind the store to one interaction instance.

    Args:
        interaction: Interaction object exposing runtime metadata helpers.
    """
    self._interaction = interaction

read ¶

read() -> SuspendedToolCall | None

Return the persisted suspended tool call from runtime metadata.

返回：

类型	描述
`SuspendedToolCall \| None`	Stored suspended tool call, or `None` when absent.

源代码位于： jianmu/tool/resume.py

def read(self) -> SuspendedToolCall | None:
    """Return the persisted suspended tool call from runtime metadata.

    Returns:
        Stored suspended tool call, or ``None`` when absent.
    """
    if self._interaction is None:
        return None
    data = self._interaction.read_runtime_metadata(SUSPENDED_TOOL_METADATA_KEY, default=None)
    return SuspendedToolCall.from_dict(data)

write ¶

write(*, action: ToolCall, request_id: str = '') -> None

Persist one suspended tool call into runtime metadata.

参数：

名称	类型	描述	默认
`action`	`ToolCall`	Tool call to persist.	必需
`request_id`	`str`	Suspension request identifier associated with the call.	`''`

源代码位于： jianmu/tool/resume.py

def write(self, *, action: ToolCall, request_id: str = "") -> None:
    """Persist one suspended tool call into runtime metadata.

    Args:
        action: Tool call to persist.
        request_id: Suspension request identifier associated with the call.
    """
    if self._interaction is None:
        return
    payload = SuspendedToolCall(
        tool=action.name,
        arguments=copy.deepcopy(action.arguments),
        tool_call_id=str(action.id or "").strip(),
        request_id=str(request_id or "").strip(),
    )
    self._interaction.write_runtime_metadata(SUSPENDED_TOOL_METADATA_KEY, payload.to_dict())

clear ¶

clear() -> None

Remove the persisted suspended tool call from runtime metadata.

源代码位于： jianmu/tool/resume.py

def clear(self) -> None:
    """Remove the persisted suspended tool call from runtime metadata."""
    if self._interaction is None:
        return
    self._interaction.clear_runtime_metadata(SUSPENDED_TOOL_METADATA_KEY)

restore_actions ¶

restore_actions(actions: list[ToolCall]) -> list[ToolCall]

Return the suspended tool call as the only action when one is persisted.

参数：

名称	类型	描述	默认
`actions`	`list[ToolCall]`	Current tool-call candidates produced by the model or runtime.	必需

返回：

类型	描述
`list[ToolCall]`	Single suspended action when one is persisted, otherwise `actions`.

源代码位于： jianmu/tool/resume.py

def restore_actions(self, actions: list[ToolCall]) -> list[ToolCall]:
    """Return the suspended tool call as the only action when one is persisted.

    Args:
        actions: Current tool-call candidates produced by the model or runtime.

    Returns:
        Single suspended action when one is persisted, otherwise ``actions``.
    """
    suspended = self.read()
    if suspended is None:
        return actions
    for action in actions:
        if suspended.tool_call_id and str(action.id or "").strip() == suspended.tool_call_id:
            return [action]
        if action.name == suspended.tool and action.arguments == suspended.arguments:
            return [action]
    return [
        ToolCall(
            name=suspended.tool,
            arguments=copy.deepcopy(suspended.arguments),
            id=suspended.tool_call_id or None,
        )
    ]

persist_active ¶

persist_active(
    *,
    actions: list[ToolCall],
    suspended_action: ToolCall | None = None,
) -> bool

Persist the current active suspension as a suspended tool call.

参数：

名称	类型	描述	默认
`actions`	`list[ToolCall]`	Current tool-call candidates associated with the suspension.	必需
`suspended_action`	`ToolCall \| None`	Explicit suspended tool call override.	`None`

返回：

类型	描述
`bool`	`True` if suspension metadata was persisted; otherwise `False`.

源代码位于： jianmu/tool/resume.py

def persist_active(self, *, actions: list[ToolCall], suspended_action: ToolCall | None = None) -> bool:
    """Persist the current active suspension as a suspended tool call.

    Args:
        actions: Current tool-call candidates associated with the suspension.
        suspended_action: Explicit suspended tool call override.

    Returns:
        ``True`` if suspension metadata was persisted; otherwise ``False``.
    """
    if self._interaction is None:
        return False
    active_suspension = self._interaction.get_active_suspension()
    if active_suspension is None:
        return False
    action = suspended_action
    if action is None and len(actions) == 1:
        action = actions[0]
    if action is not None:
        self.write(
            action=action,
            request_id=str(getattr(active_suspension, "request_id", "") or ""),
        )
    return True

Complete `dataclass` ¶

Complete(
    state_patch: Mapping[str, Any] = dict(),
    status: Status = Status.SUCCESS,
)

Complete the current ReAct loop without committing an observation.

属性：

名称	类型	描述
`state_patch`	`Mapping[str, Any]`	State updates applied before the loop completes.
`status`	`Status`	Final status propagated through ToolExecutor and ReAct.

Observe `dataclass` ¶

Observe()

Signal that a tool result should be observed and ReAct should continue.

ToolCallBatch `dataclass` ¶

ToolCallBatch(
    type: Literal["sequential", "concurrent"],
    calls: list[ToolCall],
)

One adjacent group of tool calls sharing an execution strategy.

ToolCall `dataclass` ¶

ToolCall(
    name: str,
    arguments: Any = None,
    id: Optional[str] = None,
)

Parsed tool-call request produced by an LLM response.

属性：

名称	类型	描述
`name`	`str`	Normalized tool name requested by the model.
`arguments`	`Any`	Parsed tool arguments payload, usually a dict or list.
`id`	`Optional[str]`	Optional provider-specific tool call identifier.

from_dict `staticmethod` ¶

from_dict(data: dict) -> Optional['ToolCall']

Parse a raw tool-call payload into a normalized ToolCall.

参数：

名称	类型	描述	默认
`data`	`dict`	Raw dictionary from a provider response, expected to contain `name` and optionally `arguments` and `id` keys.	必需

返回：

类型	描述
`Optional['ToolCall']`	Normalized `ToolCall` on success, `None` when `name` is absent
`Optional['ToolCall']`	or `data` is not a dict.

源代码位于： jianmu/tool/types.py

@staticmethod
def from_dict(data: dict) -> Optional["ToolCall"]:
    """Parse a raw tool-call payload into a normalized ``ToolCall``.

    Args:
        data: Raw dictionary from a provider response, expected to contain
            ``name`` and optionally ``arguments`` and ``id`` keys.

    Returns:
        Normalized ``ToolCall`` on success, ``None`` when ``name`` is absent
        or ``data`` is not a dict.
    """
    if not isinstance(data, dict):
        return None
    tool_name = data.get("name")
    args = data.get("arguments")
    if not tool_name:
        return None
    if isinstance(args, str):
        args_str = args.strip()
        if args_str.startswith("{") or args_str.startswith("["):
            try:
                args = json.loads(args_str)
            except json.JSONDecodeError:
                pass
    tool_call_id = data.get("id")
    return ToolCall(name=str(tool_name).lower(), arguments=args, id=tool_call_id)

from_list `classmethod` ¶

from_list(items: Any) -> list['ToolCall']

Normalize a raw action list into concrete ToolCall objects.

参数：

名称	类型	描述	默认
`items`	`Any`	Items to process.	必需

返回：

类型	描述
`list['ToolCall']`	The resulting list of values.

源代码位于： jianmu/tool/types.py

@classmethod
def from_list(cls, items: Any) -> list["ToolCall"]:
    """Normalize a raw action list into concrete ``ToolCall`` objects.

    Args:
        items: Items to process.

    Returns:
        The resulting list of values.
    """
    if not isinstance(items, list):
        return []
    normalized: list[ToolCall] = []
    for item in items:
        if isinstance(item, ToolCall):
            normalized.append(item)
            continue
        if isinstance(item, dict):
            extracted = cls.from_dict(item)
            if extracted:
                normalized.append(extracted)
    return normalized

to_dict ¶

to_dict() -> dict

Convert the tool call into a plain dictionary payload.

返回：

类型	描述
`dict`	Dictionary with `tool`, `arguments`, and optionally `id`.

源代码位于： jianmu/tool/types.py

def to_dict(self) -> dict:
    """Convert the tool call into a plain dictionary payload.

    Returns:
        Dictionary with ``tool``, ``arguments``, and optionally ``id``.
    """
    result = {"tool": self.name, "arguments": self.arguments}
    if self.id:
        result["id"] = self.id
    return result

ToolResult `dataclass` ¶

ToolResult(
    tool: str,
    ok: bool,
    output: Any = None,
    error: str | None = None,
)

Normalized outcome of a tool execution.

属性：

名称	类型	描述
`tool`	`str`	Name of the tool that was executed.
`ok`	`bool`	Whether the tool run completed successfully.
`output`	`Any`	Arbitrary tool output payload returned on success.
`error`	`str \| None`	Human-readable error message returned on failure.

from_raw `classmethod` ¶

from_raw(
    tool: str, raw: Any = None, *, error: str | None = None
) -> "ToolResult"

Normalize a raw tool outcome into a standard ToolResult.

参数：

名称	类型	描述	默认
`tool`	`str`	The `tool` value.	必需
`raw`	`Any`	The `raw` value.	`None`
`error`	`str \| None`	The `error` value.	`None`

返回：

类型	描述
`'ToolResult'`	The resulting `'ToolResult'` value.

源代码位于： jianmu/tool/types.py

@classmethod
def from_raw(
    cls,
    tool: str,
    raw: Any = None,
    *,
    error: str | None = None,
) -> "ToolResult":
    """Normalize a raw tool outcome into a standard ``ToolResult``.

    Args:
        tool: The `tool` value.
        raw: The `raw` value.
        error: The `error` value.

    Returns:
        The resulting `'ToolResult'` value.
    """
    if isinstance(raw, ToolResult):
        if raw.tool == tool:
            return raw
        return cls(tool=tool, ok=raw.ok, output=raw.output, error=raw.error)

    if error is not None:
        return cls(tool=tool, ok=False, output=None, error=str(error))

    if isinstance(raw, dict) and isinstance(raw.get("ok"), bool):
        payload = {
            key: value
            for key, value in raw.items()
            if key not in {"tool", "ok", "output", "error"}
        }
        if raw["ok"]:
            output = raw.get("output")
            if output is None and payload:
                output = payload
            return cls(tool=tool, ok=True, output=output, error=None)
        err = raw.get("error")
        if err is None and raw.get("output") is not None:
            err = str(raw.get("output"))
        return cls(tool=tool, ok=False, output=None, error=str(err or "tool_error"))

    if isinstance(raw, str) and raw.strip().lower().startswith("error:"):
        return cls(tool=tool, ok=False, output=None, error=raw.strip())

    return cls(tool=tool, ok=True, output=raw, error=None)

to_dict ¶

to_dict() -> dict

Convert the tool result into a plain dictionary payload.

返回：

类型	描述
`dict`	Dictionary with `tool`, `ok`, `output`, and `error` fields.

源代码位于： jianmu/tool/types.py

def to_dict(self) -> dict:
    """Convert the tool result into a plain dictionary payload.

    Returns:
        Dictionary with ``tool``, ``ok``, ``output``, and ``error`` fields.
    """
    return {
        "tool": self.tool,
        "ok": self.ok,
        "output": self.output,
        "error": self.error,
    }

FunctionTool ¶

FunctionTool(
    name: str,
    description: str,
    fn: Callable[..., Any],
    input_schema: Optional[dict] = None,
    output_schema: Optional[dict] = None,
)

Bases: Tool

Wrap a Python callable as a Tool instance.

FunctionTool is the implementation returned by @tool. It preserves the wrapped callable's name/docstring-derived metadata and can expose sync or async functions through the common Tool interface.

属性：

名称	类型	描述
`name`		Public tool name exposed to callers.
`description`		Human-readable tool description.
`_fn`		Wrapped Python callable executed by the tool.

Wrap a callable and optional schemas as a Tool object.

参数：

名称	类型	描述	默认
`name`	`str`	Public tool name exposed to agents.	必需
`description`	`str`	Human-readable tool description.	必需
`fn`	`Callable[..., Any]`	Wrapped callable.	必需
`input_schema`	`Optional[dict]`	Optional explicit JSON Schema for inputs.	`None`
`output_schema`	`Optional[dict]`	Optional explicit JSON Schema for outputs.	`None`

源代码位于： jianmu/tool/decorator.py

def __init__(
    self,
    name: str,
    description: str,
    fn: Callable[..., Any],
    input_schema: Optional[dict] = None,
    output_schema: Optional[dict] = None,
):
    """Wrap a callable and optional schemas as a ``Tool`` object.

    Args:
        name: Public tool name exposed to agents.
        description: Human-readable tool description.
        fn: Wrapped callable.
        input_schema: Optional explicit JSON Schema for inputs.
        output_schema: Optional explicit JSON Schema for outputs.
    """
    self.name = name
    self.description = description
    self._fn = fn
    if input_schema is not None:
        self.input_schema = input_schema
    if output_schema is not None:
        self.output_schema = output_schema
    if inspect.iscoroutinefunction(fn):
        async def _async_run(*args: Any, **kwargs: Any) -> Any:
            """Await the wrapped async callable."""
            return await fn(*args, **kwargs)

        self.run = _async_run  # type: ignore[method-assign]

run ¶

run(*args: Any, **kwargs: Any) -> Any

Execute the wrapped callable.

参数：

名称	类型	描述	默认
`*args`	`Any`	Positional arguments forwarded to the wrapped function.	`()`
`**kwargs`	`Any`	Keyword arguments forwarded to the wrapped function.	`{}`

返回：

类型	描述
`Any`	The return value of the wrapped function.

源代码位于： jianmu/tool/decorator.py

def run(self, *args: Any, **kwargs: Any) -> Any:
    """Execute the wrapped callable.

    Args:
        *args: Positional arguments forwarded to the wrapped function.
        **kwargs: Keyword arguments forwarded to the wrapped function.

    Returns:
        The return value of the wrapped function.
    """
    return self._fn(*args, **kwargs)

CalculatorTool ¶

Bases: Tool

Evaluate a math expression like 2+2 or 3*4-5.

属性：

名称	类型	描述
`name`		Public tool name exposed to callers.
`description`		Human-readable tool description.
`input_schema`		JSON schema describing the accepted expression input.
`output_schema`		JSON schema describing the result output.

run ¶

run(input: str) -> str

Evaluate a restricted arithmetic expression.

参数：

名称	类型	描述	默认
`input`	`str`	The arithmetic expression to evaluate.	必需

返回：

类型	描述
`str`	The evaluation result as a string, or an error message.

源代码位于： jianmu/tool/builtin/calculator.py

def run(self, input: str) -> str:
    """Evaluate a restricted arithmetic expression.

    Args:
        input: The arithmetic expression to evaluate.

    Returns:
        The evaluation result as a string, or an error message.
    """
    try:
        # Allow only math characters
        allowed_chars = set("0123456789+-*/().% ")
        if not all(c in allowed_chars for c in input):
            return f"Error: Only math expressions allowed, got: {input}"
        result = eval(input)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

PythonREPLTool ¶

PythonREPLTool(
    timeout: float = 30.0,
    max_output_length: int = 10000,
    allowed_modules: Optional[list] = None,
    globals_dict: Optional[Dict[str, Any]] = None,
)

Bases: Tool

Execute Python code and return the output.

WARNING: This tool executes arbitrary Python code. Use with caution and consider sandboxing in production environments.

属性：

名称	类型	描述
`timeout`		Maximum execution time in seconds.
`max_output_length`		Maximum output length returned to callers.
`allowed_modules`		Optional allowlist of permitted module names.
`_globals`		Global namespace used for code execution.
`_locals`	`Dict[str, Any]`	Local namespace preserved across executions.

Initialize the Python REPL tool.

参数：

名称	类型	描述	默认
`timeout`	`float`	Maximum execution time in seconds.	`30.0`
`max_output_length`	`int`	Maximum output string length.	`10000`
`allowed_modules`	`Optional[list]`	Optional list of allowed module names.	`None`
`globals_dict`	`Optional[Dict[str, Any]]`	Optional globals dictionary for code execution.	`None`

源代码位于： jianmu/tool/builtin/python_repl.py

def __init__(
    self,
    timeout: float = 30.0,
    max_output_length: int = 10000,
    allowed_modules: Optional[list] = None,
    globals_dict: Optional[Dict[str, Any]] = None,
):
    """Initialize the Python REPL tool.

    Args:
        timeout: Maximum execution time in seconds.
        max_output_length: Maximum output string length.
        allowed_modules: Optional list of allowed module names.
        globals_dict: Optional globals dictionary for code execution.
    """
    self.timeout = timeout
    self.max_output_length = max_output_length
    self.allowed_modules = allowed_modules
    self._globals = globals_dict or {"__builtins__": __builtins__}
    self._locals: Dict[str, Any] = {}

run `async` ¶

run(code: str | None = None, **kwargs: Any) -> str

Execute Python code asynchronously with timeout.

参数：

名称	类型	描述	默认
`code`	`str \| None`	The Python code snippet to execute.	`None`
`**kwargs`	`Any`	Extra parameters.	`{}`

返回：

类型	描述
`str`	The execution output (stdout/stderr or evaluation value) as a string.

源代码位于： jianmu/tool/builtin/python_repl.py

async def run(self, code: str | None = None, **kwargs: Any) -> str:
    """Execute Python code asynchronously with timeout.

    Args:
        code: The Python code snippet to execute.
        **kwargs: Extra parameters.

    Returns:
        The execution output (stdout/stderr or evaluation value) as a string.
    """
    if code is None:
        code = kwargs.get("input", "")

    if not code.strip():
        return "Error: No code provided"

    try:
        # Run in executor to avoid blocking
        loop = asyncio.get_event_loop()
        result = await asyncio.wait_for(
            loop.run_in_executor(None, self._execute_code, code),
            timeout=self.timeout
        )

        # Truncate if too long
        if len(result) > self.max_output_length:
            result = result[:self.max_output_length] + f"\n... (truncated, {len(result)} chars total)"

        return result

    except asyncio.TimeoutError:
        return f"Error: Code execution timed out after {self.timeout}s"
    except Exception as e:
        return f"Error: {type(e).__name__}: {e}"

FileReadTool ¶

FileReadTool(
    max_file_size: int = 1000000,
    allowed_extensions: Optional[list] = None,
    base_path: Optional[str] = None,
)

Bases: Tool

Read contents of a file.

属性：

名称	类型	描述
`max_file_size`		Maximum allowed file size for reads, in bytes.
`allowed_extensions`		Optional allowlist of permitted file extensions.
`base_path`		Optional root directory restricting accessible paths.

Initialize the file-read tool.

参数：

名称	类型	描述	默认
`max_file_size`	`int`	Maximum file size in bytes.	`1000000`
`allowed_extensions`	`Optional[list]`	Optional list of allowed file extensions.	`None`
`base_path`	`Optional[str]`	Optional base path used to restrict file access.	`None`

源代码位于： jianmu/tool/builtin/file.py

def __init__(
    self,
    max_file_size: int = 1_000_000,  # 1MB
    allowed_extensions: Optional[list] = None,
    base_path: Optional[str] = None,
):
    """Initialize the file-read tool.

    Args:
        max_file_size: Maximum file size in bytes.
        allowed_extensions: Optional list of allowed file extensions.
        base_path: Optional base path used to restrict file access.
    """
    self.max_file_size = max_file_size
    self.allowed_extensions = allowed_extensions
    self.base_path = Path(base_path) if base_path else None

run `async` ¶

run(
    path: str | None = None,
    encoding: str = "utf-8",
    **kwargs: Any,
) -> str

Read file contents.

参数：

名称	类型	描述	默认
`path`	`str \| None`	Path to the file.	`None`
`encoding`	`str`	Text encoding of the file.	`'utf-8'`
`**kwargs`	`Any`	Extra parameters.	`{}`

返回：

类型	描述
`str`	The text contents of the file.

源代码位于： jianmu/tool/builtin/file.py

async def run(self, path: str | None = None, encoding: str = "utf-8", **kwargs: Any) -> str:
    """Read file contents.

    Args:
        path: Path to the file.
        encoding: Text encoding of the file.
        **kwargs: Extra parameters.

    Returns:
        The text contents of the file.
    """
    if path is None:
        path = kwargs.get("input", "")

    if not path:
        raise ValueError("No path provided")

    file_path, error = self._validate_path(path)
    if error:
        self._raise_path_validation_error(error)

    loop = asyncio.get_event_loop()
    try:
        return await loop.run_in_executor(
            None,
            lambda: file_path.read_text(encoding=encoding)
        )
    except UnicodeDecodeError as exc:
        raise ValueError(f"Cannot decode file with {encoding} encoding") from exc

extract_path `staticmethod` ¶

extract_path(args: dict[str, Any] | None) -> str

Return the normalized file path from one tool-call payload.

源代码位于： jianmu/tool/builtin/file.py

@staticmethod
def extract_path(args: dict[str, Any] | None) -> str:
    """Return the normalized file path from one tool-call payload."""
    payload = dict(args or {})
    return str(payload.get("path") or payload.get("input") or "").strip()

FileWriteTool ¶

FileWriteTool(
    max_content_size: int = 1000000,
    allowed_extensions: Optional[list] = None,
    base_path: Optional[str] = None,
    create_dirs: bool = True,
)

Bases: Tool

Write contents to a file.

属性：

名称	类型	描述
`max_content_size`		Maximum allowed content size for writes, in bytes.
`allowed_extensions`		Optional allowlist of permitted file extensions.
`base_path`		Optional root directory restricting accessible paths.
`create_dirs`		Whether missing parent directories are created automatically.

Initialize the file-write tool.

参数：

名称	类型	描述	默认
`max_content_size`	`int`	Maximum content size in bytes.	`1000000`
`allowed_extensions`	`Optional[list]`	Optional list of allowed file extensions.	`None`
`base_path`	`Optional[str]`	Optional base path used to restrict file access.	`None`
`create_dirs`	`bool`	Whether missing parent directories should be created.	`True`

源代码位于： jianmu/tool/builtin/file.py

def __init__(
    self,
    max_content_size: int = 1_000_000,  # 1MB
    allowed_extensions: Optional[list] = None,
    base_path: Optional[str] = None,
    create_dirs: bool = True,
):
    """Initialize the file-write tool.

    Args:
        max_content_size: Maximum content size in bytes.
        allowed_extensions: Optional list of allowed file extensions.
        base_path: Optional base path used to restrict file access.
        create_dirs: Whether missing parent directories should be created.
    """
    self.max_content_size = max_content_size
    self.allowed_extensions = allowed_extensions
    self.base_path = Path(base_path) if base_path else None
    self.create_dirs = create_dirs

run `async` ¶

run(
    path: str | None = None,
    content: str | None = None,
    append: bool = False,
    **kwargs: Any,
) -> str

Write or append UTF-8 content to a validated file path.

参数：

名称	类型	描述	默认
`path`	`str \| None`	Path to the file.	`None`
`content`	`str \| None`	The text content to write.	`None`
`append`	`bool`	Whether to append rather than overwrite.	`False`
`**kwargs`	`Any`	Extra parameters.	`{}`

返回：

类型	描述
`str`	A status message indicating success or failure.

源代码位于： jianmu/tool/builtin/file.py

async def run(
    self, 
    path: str | None = None,
    content: str | None = None,
    append: bool = False,
    **kwargs: Any,
) -> str:
    """Write or append UTF-8 content to a validated file path.

    Args:
        path: Path to the file.
        content: The text content to write.
        append: Whether to append rather than overwrite.
        **kwargs: Extra parameters.

    Returns:
        A status message indicating success or failure.
    """
    if path is None:
        path = kwargs.get("input", "")
    if content is None:
        content = kwargs.get("text", "")

    if not path:
        return "Error: No path provided"
    if content is None:
        return "Error: No content provided"

    # Check content size
    if len(content) > self.max_content_size:
        return f"Error: Content too large ({len(content)} bytes, max: {self.max_content_size})"

    file_path, error = self._validate_path(path)
    if error:
        return f"Error: {error}"

    try:
        loop = asyncio.get_event_loop()

        def write_file():
            """Write content to disk and return the resulting file size."""
            if self.create_dirs:
                file_path.parent.mkdir(parents=True, exist_ok=True)

            mode = "a" if append else "w"
            with open(file_path, mode, encoding="utf-8") as f:
                f.write(content)
            return file_path.stat().st_size

        size = await loop.run_in_executor(None, write_file)
        action = "Appended to" if append else "Wrote"
        return f"{action} {path} ({size} bytes)"

    except Exception as e:
        return f"Error: {type(e).__name__}: {e}"

extract_path `staticmethod` ¶

extract_path(args: dict[str, Any] | None) -> str

Return the normalized file path from one tool-call payload.

源代码位于： jianmu/tool/builtin/file.py

@staticmethod
def extract_path(args: dict[str, Any] | None) -> str:
    """Return the normalized file path from one tool-call payload."""
    payload = dict(args or {})
    return str(payload.get("path") or payload.get("input") or "").strip()

HTTPTool ¶

HTTPTool(
    timeout: float = 30.0,
    max_response_length: int = 50000,
    default_headers: Optional[Dict[str, str]] = None,
    retries: int = 3,
)

Bases: Tool

Make HTTP requests to external APIs.

Supports GET, POST, PUT, DELETE methods with JSON payloads.

属性：

名称	类型	描述
`timeout`		Per-request timeout in seconds.
`max_response_length`		Maximum response-body length returned to callers.
`default_headers`		Default headers merged into every request.
`retries`		Default retry count for transient request failures.

Initialize the HTTP tool.

参数：

名称	类型	描述	默认
`timeout`	`float`	Request timeout in seconds.	`30.0`
`max_response_length`	`int`	Maximum response-body length returned to callers.	`50000`
`default_headers`	`Optional[Dict[str, str]]`	Default headers included in all requests.	`None`
`retries`	`int`	Default retry count for transient failures.	`3`

源代码位于： jianmu/tool/builtin/http.py

def __init__(
    self,
    timeout: float = 30.0,
    max_response_length: int = 50000,
    default_headers: Optional[Dict[str, str]] = None,
    retries: int = 3,
):
    """Initialize the HTTP tool.

    Args:
        timeout: Request timeout in seconds.
        max_response_length: Maximum response-body length returned to callers.
        default_headers: Default headers included in all requests.
        retries: Default retry count for transient failures.
    """
    self.timeout = timeout
    self.max_response_length = max_response_length
    self.default_headers = default_headers or {}
    self.retries = retries

parallel_decision_for_call ¶

parallel_decision_for_call(
    args: Any, *, runtime_injected: bool
) -> ParallelDecision

Allow concurrent HTTP calls only when their method is read-only.

源代码位于： jianmu/tool/builtin/http.py

def parallel_decision_for_call(self, args: Any, *, runtime_injected: bool) -> ParallelDecision:
    """Allow concurrent HTTP calls only when their method is read-only."""
    if runtime_injected:
        return super().parallel_decision_for_call(args, runtime_injected=runtime_injected)
    payload = args if isinstance(args, dict) else {}
    method = str(payload.get("method") or "GET").strip().upper()
    if method in {"GET", "HEAD"}:
        return ParallelDecision(
            allowed=True,
            reason=f"HTTP {method} is treated as read-only",
            category="allowed",
        )
    return ParallelDecision(
        allowed=False,
        reason=f"HTTP {method} may mutate remote state",
        category="args_sensitive",
    )

run `async` ¶

run(
    url: str | None = None,
    method: str = "GET",
    headers: Optional[Dict[str, str]] = None,
    body: Optional[Dict[str, Any]] = None,
    retries: Optional[int] = None,
    **kwargs: Any,
) -> str

Make an HTTP request.

参数：

名称	类型	描述	默认
`url`	`str \| None`	The destination URL.	`None`
`method`	`str`	HTTP method (e.g. GET, POST).	`'GET'`
`headers`	`Optional[Dict[str, str]]`	Optional dictionary of HTTP headers.	`None`
`body`	`Optional[Dict[str, Any]]`	Optional dictionary to send as JSON body.	`None`
`retries`	`Optional[int]`	Number of retry attempts.	`None`
`**kwargs`	`Any`	Extra parameters.	`{}`

返回：

类型	描述
`str`	The HTTP response status and body as a string.

源代码位于： jianmu/tool/builtin/http.py

async def run(
    self,
    url: str | None = None,
    method: str = "GET",
    headers: Optional[Dict[str, str]] = None,
    body: Optional[Dict[str, Any]] = None,
    retries: Optional[int] = None,
    **kwargs: Any,
) -> str:
    """Make an HTTP request.

    Args:
        url: The destination URL.
        method: HTTP method (e.g. GET, POST).
        headers: Optional dictionary of HTTP headers.
        body: Optional dictionary to send as JSON body.
        retries: Number of retry attempts.
        **kwargs: Extra parameters.

    Returns:
        The HTTP response status and body as a string.
    """
    # Handle input from various formats
    if url is None:
        url = kwargs.get("input", "")

    if not url:
        return "Error: No URL provided"

    method = method.upper()
    merged_headers = {**self.default_headers, **(headers or {})}

    attempts = retries if retries is not None else self.retries
    last_error: Optional[str] = None
    for attempt in range(1, max(1, attempts) + 1):
        try:
            timeout = aiohttp.ClientTimeout(total=self.timeout)
            async with aiohttp.ClientSession(timeout=timeout) as session:
                request_kwargs = {"headers": merged_headers}

                if body and method in ("POST", "PUT"):
                    request_kwargs["json"] = body

                async with session.request(method, url, **request_kwargs) as response:
                    status = response.status
                    content_type = response.headers.get("Content-Type", "")

                    if "application/json" in content_type:
                        try:
                            data = await response.json()
                            text = json.dumps(data, ensure_ascii=False, indent=2)
                        except Exception:
                            text = await response.text()
                    else:
                        text = await response.text()

                    if len(text) > self.max_response_length:
                        text = text[:self.max_response_length] + "\n... (truncated)"

                    return f"[HTTP {status}]\n{text}"

        except asyncio.TimeoutError:
            last_error = f"Error: Request timed out after {self.timeout}s"
        except aiohttp.ClientError as e:
            last_error = f"Error: {type(e).__name__}: {e}"
        except Exception as e:
            last_error = f"Error: {type(e).__name__}: {e}"

        if attempt < attempts:
            await asyncio.sleep(0)

    return last_error or "Error: Request failed"

DuckDuckGoSearchTool ¶

DuckDuckGoSearchTool()

Bases: Tool

DuckDuckGo web search (no API key required).

属性：

名称	类型	描述
`name`		Public tool name exposed to callers.
`description`		Human-readable tool description.
`_ddgs_cls`		Lazily imported DuckDuckGo client class.

Resolve the DuckDuckGo client dependency lazily.

源代码位于： jianmu/tool/builtin/duckduckgo.py

def __init__(self):
    """Resolve the DuckDuckGo client dependency lazily."""
    self._ddgs_cls = _resolve_ddgs_cls()

run `async` ¶

run(query: str, max_results: int = 5) -> str

Execute search via DDG.

参数：

名称	类型	描述	默认
`query`	`str`	The search query string.	必需
`max_results`	`int`	The maximum number of search results to return.	`5`

返回：

类型	描述
`str`	The search results formatted as a string.

源代码位于： jianmu/tool/builtin/duckduckgo.py

async def run(self, query: str, max_results: int = 5) -> str:
    """Execute search via DDG.

    Args:
        query: The search query string.
        max_results: The maximum number of search results to return.

    Returns:
        The search results formatted as a string.
    """
    logger.debug(f"🔍 [DuckDuckGo] Searching for: {query}")

    try:
        # Use run_in_executor if the library is not async, 
        # but ddgs has a synchronous and asynchronous version. 
        # We'll use the sync one within an executor for safety if preferred,
        # or just use the async version if available.
        # duckduckgo_search 6.x+ has DDGS context manager.

        loop = asyncio.get_event_loop()
        results = await loop.run_in_executor(None, self._sync_search, query, max_results)

        if not results:
            return "No results found for this query."

        formatted_results = []
        for i, r in enumerate(results, 1):
            formatted_results.append(
                f"[{i}] {r.get('title')}\n"
                f"URL: {r.get('href')}\n"
                f"Snippet: {r.get('body')}\n"
            )

        return "\n".join(formatted_results)

    except Exception as e:
        logger.error(f"❌ [DuckDuckGo] Search failed: {e}")
        return f"Error performing search: {str(e)}"

tool ¶

tool(
    _func: Optional[Callable[..., Any]] = None,
    *,
    name: Optional[str] = None,
    description: Optional[str] = None,
    input_schema: Optional[dict] = None,
    output_schema: Optional[dict] = None,
) -> (
    FunctionTool
    | Callable[[Callable[..., Any]], FunctionTool]
)

Decorator to wrap a function into a Tool instance.

Can be used with or without arguments::

@tool
def my_tool(input): ...

@tool(name="custom")
def my_tool(input): ...

The wrapped function becomes a first-class Tool object, so it can be passed directly into ToolExecutor or converted to JSON tool schema for model providers.

参数：

名称	类型	描述	默认
`_func`	`Optional[Callable[..., Any]]`	The function to wrap when used without parentheses.	`None`
`name`	`Optional[str]`	Optional tool name override (defaults to the function name).	`None`
`description`	`Optional[str]`	Optional description override (defaults to docstring).	`None`
`input_schema`	`Optional[dict]`	Optional JSON Schema for the tool input.	`None`
`output_schema`	`Optional[dict]`	Optional JSON Schema for the tool output.	`None`

返回：

类型	描述
`FunctionTool \| Callable[[Callable[..., Any]], FunctionTool]`	A `FunctionTool` instance (direct decoration) or a decorator
`FunctionTool \| Callable[[Callable[..., Any]], FunctionTool]`	that produces one.

源代码位于： jianmu/tool/decorator.py

def tool(
    _func: Optional[Callable[..., Any]] = None,
    *,
    name: Optional[str] = None,
    description: Optional[str] = None,
    input_schema: Optional[dict] = None,
    output_schema: Optional[dict] = None,
) -> FunctionTool | Callable[[Callable[..., Any]], FunctionTool]:
    """Decorator to wrap a function into a Tool instance.

    Can be used with or without arguments::

        @tool
        def my_tool(input): ...

        @tool(name="custom")
        def my_tool(input): ...

    The wrapped function becomes a first-class ``Tool`` object, so it can be
    passed directly into ``ToolExecutor`` or converted to JSON tool schema for
    model providers.

    Args:
        _func: The function to wrap when used without parentheses.
        name: Optional tool name override (defaults to the function name).
        description: Optional description override (defaults to docstring).
        input_schema: Optional JSON Schema for the tool input.
        output_schema: Optional JSON Schema for the tool output.

    Returns:
        A ``FunctionTool`` instance (direct decoration) or a decorator
        that produces one.
    """
    def decorator(func: Callable[..., Any]) -> FunctionTool:
        """Wrap a function as a ``FunctionTool`` instance.

        Args:
            func: The function to wrap.

        Returns:
            A ``FunctionTool`` wrapping the function.
        """
        tool_name, tool_desc = _get_metadata(func, name, description)
        return FunctionTool(
            name=tool_name,
            description=tool_desc,
            fn=func,
            input_schema=input_schema,
            output_schema=output_schema,
        )

    if _func is None:
        return decorator
    return decorator(_func)

jianmu.tool¶

1. 模块职责¶

2. 适合查什么¶

3. 注意事项¶

4. builtin 路径说明¶

5. 最小示例¶

6. 常见入口¶

7. API 参考¶

tool ¶

ParallelDecision dataclass ¶

Tool ¶

run abstractmethod ¶

execute async ¶

permission_check_for_call ¶

parallel_decision_for_call ¶

as_node ¶

get_description ¶

get_effect_tags ¶

to_schema ¶

spec ¶

ToolSet dataclass ¶

from_tools classmethod ¶

resolve classmethod ¶

from_providers classmethod ¶

list_builtin_names classmethod ¶

merge ¶

with_tools ¶

get ¶

execute async ¶

is_parallel_safe ¶

schemas ¶

specs ¶

describe ¶

names ¶

BuiltinToolProvider ¶

aliases classmethod ¶

register_tool classmethod ¶

register_alias classmethod ¶

canonical_tool_name classmethod ¶

list_builtin_names classmethod ¶

is_builtin_tool_name classmethod ¶

resolve_named_tools classmethod ¶

get_tools classmethod ¶

ToolProvider ¶

initialize async ¶

get_tools ¶

close async ¶

BaseToolProvider ¶

initialize async ¶

close async ¶

SuspendedToolCall dataclass ¶

from_dict classmethod ¶

to_dict ¶

SuspendedToolCallStore ¶

read ¶

write ¶

clear ¶

restore_actions ¶

persist_active ¶

Complete dataclass ¶

Observe dataclass ¶

ToolCallBatch dataclass ¶

ToolCall dataclass ¶

from_dict staticmethod ¶

from_list classmethod ¶

to_dict ¶

ToolResult dataclass ¶

from_raw classmethod ¶

to_dict ¶

FunctionTool ¶

run ¶

CalculatorTool ¶

run ¶

PythonREPLTool ¶

run async ¶

FileReadTool ¶

run async ¶

extract_path staticmethod ¶

FileWriteTool ¶

run async ¶

`jianmu.tool`¶

ParallelDecision `dataclass` ¶

run `abstractmethod` ¶

execute `async` ¶

ToolSet `dataclass` ¶

from_tools `classmethod` ¶

resolve `classmethod` ¶

from_providers `classmethod` ¶

list_builtin_names `classmethod` ¶

execute `async` ¶

aliases `classmethod` ¶

register_tool `classmethod` ¶

register_alias `classmethod` ¶

canonical_tool_name `classmethod` ¶

list_builtin_names `classmethod` ¶

is_builtin_tool_name `classmethod` ¶

resolve_named_tools `classmethod` ¶

get_tools `classmethod` ¶

initialize `async` ¶

close `async` ¶

initialize `async` ¶

close `async` ¶

SuspendedToolCall `dataclass` ¶

from_dict `classmethod` ¶

Complete `dataclass` ¶

Observe `dataclass` ¶

ToolCallBatch `dataclass` ¶

ToolCall `dataclass` ¶

from_dict `staticmethod` ¶

from_list `classmethod` ¶

ToolResult `dataclass` ¶

from_raw `classmethod` ¶

run `async` ¶

run `async` ¶

extract_path `staticmethod` ¶

run `async` ¶

extract_path `staticmethod` ¶

run `async` ¶

run `async` ¶