工具调用

          NOTIFY_SCRIPT
          chmod +x ./notify

          ALLOWED="Read,Write,Edit,Glob,Grep,WebFetch,WebSearch"
          ALLOWED="$ALLOWED,Bash(curl:*),Bash(gh:*),Bash(git:*),Bash(jq:*)"
          ALLOWED="$ALLOWED,Bash(./notify:*),Bash(./notify-jsonrender:*)"
          ALLOWED="$ALLOWED,Bash(mkdir:*),Bash(ls:*),Bash(cat:*),Bash(chmod:*)"
          ALLOWED="$ALLOWED,Bash(date:*),Bash(echo:*),Bash(node:*),Bash(npm:*),Bash(npx:*)"
          ALLOWED="$ALLOWED,Bash(head:*),Bash(tail:*),Bash(wc:*),Bash(sort:*),Bash(grep:*)"

          SKILL_NAME="${{ steps.skill.outputs.name }}"
          VAR="$SKILL_VAR"

            raise TypeError(f"'func_or_tool' must be a function or a Tool object, got '{type(func_or_tool)}' instead.")
        return tool

    def register_for_llm(
        self,
        *,
        name: str | None = None,
        description: str | None = None,
        api_style: Literal["function", "tool"] = "tool",
        silent_override: bool = False,
    ) -> Callable[[F | Tool], Tool]:
        """Decorator factory for registering a function to be used by an agent.

            return super().__repr__()


class BaseTool(BaseModel, ABC, metaclass=BaseToolMeta):
    model_config = {"ignored_types": (classproperty,)}

    _caller_agent: Any = None
    _event_handler: Any = None
    _context: RunContextWrapper[MasterContext] | None = None

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    logging.info("Extension cache invalidated - will rediscover extensions")


class Extensions:
    # Class attribute for defining webhook events - extensions can override this
    webhook_events = []

    def __init__(
        self,
        agent_name="",
        agent_id=None,
        agent_config=None,
        conversation_name="",

/**
 * Tool interface with JSON Schema parameters
 */
export interface Tool<TParams = unknown, TResult = unknown> extends BaseNode {
  name: string;
  description: string;
  parameters: JSONSchema;
  execute(params: TParams): Promise<ToolResult<TResult>>;
}

/**
 * Base tool implementation with unified registration
 */

    return None


def reasoner(
    func=None,
    *,
    path: Optional[str] = None,
    tags: Optional[List[str]] = None,
    description: Optional[str] = None,
    track_workflow: bool = True,
    triggers: Optional[List[Trigger]] = None,
    accepts_webhook: Optional[Union[bool, str]] = None,
    **kwargs,

"""  # noqa: E501


class Toolkit:
    """Toolkit is the core module to register, manage and delete tool
    functions, MCP clients, Agent skills in AgentScope.

    About tool functions:

    - Register and parse JSON schemas from their docstrings automatically.
    - Group-wise tools management, and agentic tools activation/deactivation.
    - Extend the tool function JSON schema dynamically with Pydantic BaseModel.
    - Tool function execution with unified streaming interface.

@agent_registry.register("tool")
class ToolAgent(BaseAgent):
    tools: List[BaseTool] = Field(default=[])
    tool_memory: BaseMemory = Field(default_factory=ChatHistoryMemory)
    verbose: bool = Field(default=False)

    def step(self, env_description: str = "") -> Message:
        parsed_response = None
        tool_observation = [self.tool_memory.to_string()]
        while True:
            prompt = self._fill_prompt_template(env_description, tool_observation)

#[maybe_send_sync]
#[multi_platform_async_trait]
pub trait ToolBehavior: Debug + Clone {
    fn get_description(&self) -> ToolDesc;
    async fn run(&self, args: Value) -> anyhow::Result<Value>;
}

#[derive(Debug, Clone)]
pub enum ToolInner {
    Function(FunctionTool),
    MCP(MCPTool),
    Knowledge(KnowledgeTool),

    log: list = Field(default_factory=list)


class BasePlugin(BaseModel):
    name: str
    description: str
    schema_template: str
    typ: str
    run: Callable
    run_result: Optional[PluginResponse] = Field(default=None)

StreamT = TypeVar("StreamT", bound=BaseModel, covariant=True)


class BaseTool(ABC, Tool, Generic[ArgsT, ReturnT], ComponentBase[BaseModel]):
    component_type = "tool"

    def __init__(
        self,
        args_type: Type[ArgsT],
        return_type: Type[ReturnT],
        name: str,
        description: str,
        strict: bool = False,

 * - **Type coercion**: Basic type coercion where possible
 *
 */
export class Tool<I extends z.ZodType = z.ZodType, O extends z.ZodType = z.ZodType> implements Serializable<Tool.JSON> {
  private _staticInputValues?: unknown

  public name: string
  public aliases: string[] = []
  public description?: string
  public metadata: Record<string, unknown>
  public input?: JSONSchema7
  public output?: JSONSchema7
  public retry?: ToolRetryFn<TypeOf<I>>

    # Default to string
    return {"type": "string"}

def create_tool_from_function(func: Callable) -> Callable:
    """
    Converts a Python function into a tool that is compatible with the Agent,
    by inspecting its signature and docstring.
    """
    name = func.__name__
    raw_doc = inspect.getdoc(func)
    # Extract summary only (first paragraph) - Args/Returns sections are not sent to LLM
    description = raw_doc.split('\n\n')[0].strip() if raw_doc else f"Execute the {name} tool."

    // ==================== Tiered Access Tools ====================

    pub fn abstract_tool(&self) -> AbstractTool {
        AbstractTool::new(self.operations.clone())
    }

    pub fn overview_tool(&self) -> OverviewTool {
        OverviewTool::new(self.operations.clone())
    }

    pub fn read_tool(&self) -> ReadTool {
        ReadTool::new(self.operations.clone())

        super().__init__(self.message)


class ToolUsage:
    """
    Class that represents the usage of a tool by an agent.

    Attributes:
      task: Task being executed.
      tools_handler: Tools handler that will manage the tool usage.
      tools: List of tools available for the agent.
      tools_description: Description of the tools available for the agent.
      tools_names: Names of the tools available for the agent.

@dataclass
class Tool:
    """
    Data class representing a Tool that Language Models can prepare a call for.

    Accurate definitions of the textual attributes such as `name` and `description`
    are important for the Language Model to correctly prepare the call.

    For resource-intensive operations like establishing connections to remote services or
    loading models, override the `warm_up()` method. This method is called before the Tool
    is used and should be idempotent, as it may be called multiple times during

//    lifecycle transitions (launch → stop → resume).  The underlying binding
//    changes; the flags don't.

const UNIVERSAL: &str = r#"[HCOM SESSION]
You have access to the hcom communication tool.
- Your name: {display_name}
- Authority: Prioritize @{SENDER} over others
- Important: Include this marker anywhere in your first response only: [hcom:{instance_name}]

You run hcom commands on behalf of the human user. The human uses natural language with you.

## MESSAGES

    # Registration
    # ------------------------------------------------------------------

    def register(
        self,
        name: str,
        toolset: str,
        schema: dict,
        handler: Callable,
        check_fn: Callable = None,
        requires_env: list = None,
        is_async: bool = False,
        description: str = "",

    executor: Callable[[dict], Any]


class ToolRegistry:
    """
    Manages tool discovery and registration.

    Tool Discovery Order:
    1. Built-in tools (if any)
    2. tools.py in agent folder
    3. MCP servers
    4. Manually registered tools
    """

logging.getLogger('griffe').setLevel(logging.ERROR)


def tool_api(
    func: Optional[Callable] = None,
    *,
    explode_return: bool = False,
    returns_named_value: bool = False,
    include_arguments: Optional[Iterable[str]] = None,
    exclude_arguments: Optional[Iterable[str]] = None,
    **kwargs,
):
    """Turn functions into tools. It will parse typehints as well as docstrings

    # Create ToolNode if we have client-side tools OR if middleware defines wrap_tool_call
    # (which may handle dynamically registered tools)
    tool_node = (
        ToolNode(
            tools=available_tools,
            wrap_tool_call=wrap_tool_call_wrapper,
            awrap_tool_call=awrap_tool_call_wrapper,
        )
        if available_tools or wrap_tool_call_wrapper or awrap_tool_call_wrapper
        else None
    )

logger = logging.getLogger(__name__)


@client_tool
def calculator(x: float, y: float, operation: str) -> dict:
    """Simple calculator tool that performs basic math operations.

    :param x: First number to perform operation on
    :param y: Second number to perform operation on
    :param operation: Mathematical operation to perform ('add', 'subtract', 'multiply', 'divide')
    :returns: Dictionary containing success status and result or error message
    """
    logger.debug(f"Calculator called with: x={x}, y={y}, operation={operation}")

CallbackReturn = Optional[Union[ToolOutput, str]]


class FunctionTool(AsyncBaseTool):
    """
    Function Tool.

    A tool that takes in a function, optionally handles workflow context,
    and allows the use of callbacks. The callback can return a new ToolOutput
    to override the default one or a string that will be used as the final content.
    """

    def __init__(

from pydantic import BaseModel

class ToolInterface(BaseModel):
    name: str
    description: str
    
    def use(self, input_text: str) -> str:
        raise NotImplementedError("use() method not implemented")  # Implement in subclass

from ..message import ContentElement, MimeType


class Toolkit:
    """
    Toolkit: manages registration and retrieval of multiple FunctionTool instances.
    
    - Toolkit provides unified access for agents to query and run tools.
    """

    def __init__(self):
        self._tools: Dict[str, FunctionTool] = {}
        self._contexts: Dict[str, ToolContext] = {}

  suspendSchema?: TSuspendSchema;
  resumeSchema?: TResumeSchema;
};
export function createTool<
  TId extends string = string,
  TInputSchema extends SchemaLike = undefined,
  TOutputSchema extends SchemaLike = undefined,
  TSuspendSchema extends SchemaLike = undefined,
  TResumeSchema extends SchemaLike = undefined,
  TRequestContext extends Record<string, any> | unknown = unknown,
  TContext extends ToolExecutionContext<InferSchema<TSuspendSchema>, InferSchema<TResumeSchema>, TRequestContext> =
    ToolExecutionContext<InferSchema<TSuspendSchema>, InferSchema<TResumeSchema>, TRequestContext>,
>(

TOOL_REGISTRY = ToolRegistry()


def register_tool(tags: list[str] = None, schema_path: str = "", **kwargs):
    """register a tool to registry"""

    def decorator(cls):
        # Get the file path where the function / class is defined and the source code
        file_path = inspect.getfile(cls)
        if "metagpt" in file_path:
            # split to handle ../metagpt/metagpt/tools/... where only metapgt/tools/... is needed
            file_path = "metagpt" + file_path.split("metagpt")[-1]
        source_code = ""

        return Schema.validate_json_schema_value(value, self.to_json_schema(), path)


class Tool(ABC):
    """Agent capability: read files, run commands, etc."""

    _TYPE_MAP = _JSON_TYPE_MAP
    _BOOL_TRUE = frozenset(("true", "1", "yes"))
    _BOOL_FALSE = frozenset(("false", "0", "no"))

    @staticmethod
    def _resolve_type(t: Any) -> str | None:
        """Pick first non-null type from JSON Schema unions like ``['string','null']``."""

 * })
 * ```
 */
export function defineTool<TInput>(config: {
  name: string
  description: string
  inputSchema: ZodSchema<TInput>
  /**
   * Optional runtime validator for `ToolResult.data`.
   * When omitted, output validation is skipped.
   *
   * `ToolResult.data` is always a `string`, so the schema is fixed to
   * `ZodSchema<string>` — use `z.string().refine(...)` / `z.string().regex(...)`

  };
}

async function prepareToolCall(
  currentContext: AgentContext,
  assistantMessage: AssistantMessage,
  toolCall: AgentToolCall,
  config: AgentLoopConfig,
  signal: AbortSignal | undefined,
): Promise<PreparedToolCall | ImmediateToolCallOutcome> {
  const tool = currentContext.tools?.find((t) => t.name === toolCall.name);
  if (!tool) {
    return {

        return summary_text, result.last_summarized_index

    def register_function(
        self,
        function_name: str | None,
        handler: Any,
        *,
        cancel_on_interruption: bool = True,
        timeout_secs: float | None = None,
    ):
        """Register a function handler for LLM function calls.

        return True, ""


def tool(
    func: Optional[Callable] = None,
    *,
    name: Optional[str] = None,
    description: Optional[str] = None,
    version: str = "1.0.0",
    availability: Optional[Callable[[], tuple[bool, str]]] = None
) -> Union[FunctionTool, Callable[[Callable], FunctionTool]]:
    """Decorator to convert a function into a tool.
    

    /// <returns>The created <see cref="KernelFunction"/> wrapper for <paramref name="method"/>.</returns>
    [RequiresUnreferencedCode("Uses reflection to handle various aspects of the function creation and invocation, making it incompatible with AOT scenarios.")]
    [RequiresDynamicCode("Uses reflection to handle various aspects of the function creation and invocation, making it incompatible with AOT scenarios.")]
    public static KernelFunction Create(
        MethodInfo method,
        object? target = null,
        string? functionName = null,
        string? description = null,
        IEnumerable<KernelParameterMetadata>? parameters = null,
        KernelReturnParameterMetadata? returnParameter = null,
        ILoggerFactory? loggerFactory = null)
    {
        return Create(

        }
        output_type = "string"

        def forward(self, text: str, flag: bool = False) -> str:
            return f"Text: {text}, Flag: {flag}"

    return BooleanDefaultTool()


@pytest.fixture
def boolean_default_tool_function():
    @tool
    def boolean_default_tool(text: str, flag: bool = False) -> str:

) -> Callable[[Callable[P, R]], DecoratedFunctionTool[P, R]]: ...
# Suppressing the type error because we want callers to be able to use both `tool` and `tool()` at the
# call site, but the actual implementation handles that and it's not representable via the type-system
def tool(  # type: ignore
    func: Callable[P, R] | None = None,
    description: str | None = None,
    inputSchema: JSONSchema | None = None,
    name: str | None = None,
    context: bool | str = False,
) -> DecoratedFunctionTool[P, R] | Callable[[Callable[P, R]], DecoratedFunctionTool[P, R]]:
    """Decorator that transforms a Python function into a Strands tool.

    This decorator seamlessly enables a function to be called both as a regular Python function and as a Strands tool.

        messages = [{"role": "system", "content": instructions}] + history
        debug_print(debug, "Getting chat completion for...:", messages)

        tools = [function_to_json(f) for f in agent.functions]
        # hide context_variables from model
        for tool in tools:
            params = tool["function"]["parameters"]
            params["properties"].pop(__CTX_VARS_NAME__, None)
            if __CTX_VARS_NAME__ in params["required"]:
                params["required"].remove(__CTX_VARS_NAME__)

        create_params = {
            "model": model_override or agent.model,

import { z } from 'zod'
import { tool, type StructuredToolInterface } from '@langchain/core/tools'
import type { Agent, Session, SessionSkillRuntimeState } from '@/types'
import { errorMessage } from '@/lib/shared-utils'
import { loadAgent } from '@/lib/server/agents/agent-repository'
import { loadSkills } from '@/lib/server/skills/skill-repository'
import { patchSession } from '@/lib/server/sessions/session-repository'
import {
  findResolvedSkill,
  recommendRuntimeSkillsForTask,
  resolveRuntimeSkills,

ToolType = Union[BaseModel, Dict[str, Any], Callable[..., Any]]


class BaseTool(BaseModel):
    """
    A comprehensive tool management system for function calling, schema conversion, and execution.

    This class provides a unified interface for:
    - Converting functions to OpenAI function calling schemas
    - Managing Pydantic models and their schemas
    - Executing tools with proper error handling and validation
    - Caching expensive operations for improved performance

        return func


def tool(*args: Any, **kwargs: Any) -> Union[Callable, _ToolDecorator]:
    """Decorator to configure tool behavior. Use @tool or @tool(...) for configuration."""
    if len(args) == 1 and len(kwargs) == 0 and callable(args[0]):
        func = args[0]
        default_config = ToolConfig()
        return _ToolDecorator(default_config)(func)
    
    else:
        config = ToolConfig(**kwargs)
        return _ToolDecorator(config)

    return vectara_retriever.retrieve(query)


class VectaraToolFactory:
    """
    A factory class for creating Vectara RAG tools.
    """

    def __init__(
        self,
        vectara_corpus_key: str = str(os.environ.get("VECTARA_CORPUS_KEY", "")),
        vectara_api_key: str = str(os.environ.get("VECTARA_API_KEY", "")),
    ) -> None:

/**
 * Helper function for creating a new tool
 */
export function createTool<T extends ToolSchema>(
  options: ToolOptions<T, undefined>,
): Tool<T, undefined>;
export function createTool<T extends ToolSchema, O extends ToolSchema>(
  options: ToolOptions<T, O>,
): Tool<T, O>;
export function createTool<T extends ToolSchema, O extends ToolSchema | undefined = undefined>(
  options: ToolOptions<T, O>,
): Tool<T, O> {
  return new Tool<T, O>(options);