Tools¶

Tool (容器 — 一组相关功能)
  +-- Function (Schema + 入口点)
  |     +-- name, description
  |     +-- parameters (JSON Schema，自动从类型注解生成)
  |     +-- entrypoint (实际调用的函数)
  |     +-- concurrency_safe, is_read_only, is_destructive
  +-- FunctionCall (单次调用实例)
        +-- args, result, error, timing

from agentica import Agent, ZhipuAI

def get_weather(city: str, unit: str = "celsius") -> str:
    """获取指定城市的实时天气

    Args:
        city: 城市名称，如 "北京"、"上海"、"New York"
        unit: 温度单位，"celsius"（摄氏）或 "fahrenheit"（华氏）
    """
    # 实际场景中调用天气 API
    return f"{city}: 晴天，22°C"

agent = Agent(model=ZhipuAI(), tools=[get_weather])
result = agent.run_sync("上海今天天气怎么样？")
print(result.content)

import aiohttp
from agentica import Agent, ZhipuAI

async def fetch_stock_price(symbol: str) -> str:
    """获取股票实时价格

    Args:
        symbol: 股票代码，如 "AAPL"、"000001.SZ"
    """
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/stock/{symbol}") as resp:
            data = await resp.json()
            return f"{symbol}: {data['price']} USD"

agent = Agent(model=ZhipuAI(), tools=[fetch_stock_price])

from agentica.tools.decorators import tool
from agentica import Agent, ZhipuAI

@tool(
    name="search_code",
    description="在代码库中搜索指定模式",
    concurrency_safe=True,    # 可与其他只读工具并发执行
    is_read_only=True,        # 声明为只读，不修改状态
)
def search_in_codebase(pattern: str, path: str = ".") -> str:
    """在代码库中搜索指定模式

    Args:
        pattern: 搜索的正则表达式或字符串
        path: 搜索的目录路径
    """
    import subprocess
    result = subprocess.run(["rg", pattern, path], capture_output=True, text=True)
    return result.stdout or "No matches found"

@tool(
    name="delete_file",
    is_destructive=True,      # 声明为破坏性操作，提示 Agent 谨慎使用
    stop_after_tool_call=True # 执行后立即停止 Agent（用于危险操作）
)
def delete_file(path: str) -> str:
    """删除指定文件（不可恢复）

    Args:
        path: 要删除的文件路径
    """
    import os
    os.remove(path)
    return f"Deleted: {path}"

agent = Agent(model=ZhipuAI(), tools=[search_in_codebase, delete_file])

from agentica import Tool, Agent, ZhipuAI

class DatabaseTool(Tool):
    """数据库查询工具，支持 SELECT 和 INSERT"""

    def __init__(self, connection_string: str):
        super().__init__(name="database")
        self.conn_str = connection_string
        # 注册工具函数，并声明元数据
        self.register(self.query, is_read_only=True, concurrency_safe=True)
        self.register(self.insert, is_destructive=True)
        self.register(self.list_tables, is_read_only=True, concurrency_safe=True)

    def query(self, sql: str, limit: int = 100) -> str:
        """执行 SQL 查询并返回结果

        Args:
            sql: SELECT 语句
            limit: 最多返回的行数
        """
        # 实际场景中执行数据库查询
        return f"Query results for: {sql}"

    def insert(self, table: str, data: dict) -> str:
        """向表中插入一行数据

        Args:
            table: 表名
            data: 要插入的数据字典
        """
        return f"Inserted into {table}: {data}"

    def list_tables(self) -> str:
        """列出数据库中所有表"""
        return "users, orders, products, inventory"

    def get_system_prompt(self) -> str:
        """可选：注入工具使用指南到 System Prompt"""
        return """## Database Tool Guidelines
- Always use SELECT before INSERT to check existing data
- Limit SELECT queries to 100 rows unless user needs more
- Never execute DELETE or DROP statements
"""

agent = Agent(
    model=ZhipuAI(),
    tools=[DatabaseTool("sqlite:///app.db")]
)

LLM 返回: [read_file("a.py"), read_file("b.py"), grep("TODO")]
          ↓
          asyncio.gather() 并发执行（3 个工具同时跑）
          ↓
结果全部就绪后，一起追加到消息历史，继续 LLM 调用

# 这两个工具可以并发：
self.register(self.read_config, concurrency_safe=True, is_read_only=True)
self.register(self.read_schema, concurrency_safe=True, is_read_only=True)

# 这个工具必须串行（写操作）：
self.register(self.write_output)  # concurrency_safe=False (默认)

from agentica.tools.base import StopAgentRun, RetryAgentRun

def validate_input(query: str) -> str:
    """验证并处理用户输入

    Args:
        query: 用户查询内容
    """
    if not query.strip():
        # 告诉 Agent 重新考虑，用 message 作为反馈
        raise RetryAgentRun("query 不能为空，请重新提供有意义的查询内容")

    if any(word in query.lower() for word in ["delete", "drop", "truncate"]):
        # 立即终止整个 Agent 执行
        raise StopAgentRun("检测到高危 SQL 关键字，拒绝执行")

    return f"处理结果: {query}"

from typing import Literal, Optional, List

def process_data(
    input_text: str,           # → {"type": "string"}
    max_items: int = 10,       # → {"type": "number"}
    include_meta: bool = False,# → {"type": "boolean"}
    format: Literal["json", "csv", "txt"] = "json",  # → {"type": "string", "enum": ["json","csv","txt"]}
    tags: Optional[List[str]] = None,  # → {"type": "array", "items": {"type": "string"}}
) -> str:
    """处理数据并返回结果"""
    ...

{
  "name": "process_data",
  "description": "处理数据并返回结果",
  "parameters": {
    "type": "object",
    "properties": {
      "input_text": {"type": "string"},
      "max_items": {"type": "number"},
      "include_meta": {"type": "boolean"},
      "format": {"type": "string", "enum": ["json", "csv", "txt"]},
      "tags": {"type": "array", "items": {"type": "string"}}
    },
    "required": ["input_text"]
  }
}

from agentica.tools.builtin import get_builtin_tools

# 获取全套内置工具
tools = get_builtin_tools(work_dir="./")

# SDK：显式指定
agent = Agent(tools=get_builtin_tools(web_search_provider="bocha"))

# SDK：直接构造
agent = Agent(tools=[BuiltinWebSearchTool(provider="exa"), ...])

# CLI + SDK 通用：环境变量
export AGENTICA_WEB_SEARCH=bocha
export BOCHA_API_KEY=sk-xxx

env:
  AGENTICA_WEB_SEARCH: bocha
  BOCHA_API_KEY: "sk-..."

export AGENTICA_WEB_SEARCH=zhipu
export ZAI_API_KEY=sk-xxx
export AGENTICA_ZHIPU_SEARCH_ENGINE=search_pro_sogou   # 需要每条都有来源 URL 时

from agentica.tools.zhipu_web_search_tool import ZhipuWebSearchTool

tool = ZhipuWebSearchTool(
    search_engine="search_pro",
    search_recency_filter="oneWeek",   # oneDay / oneWeek / oneMonth / oneYear / noLimit
    search_domain_filter="www.sohu.com",
    content_size="high",              # medium（默认）/ high，high 摘要长度约翻倍
    user_id="tenant-000042",          # 6-128 字符，一个 key 服务多个终端用户时传
)

export AGENTICA_WEB_SEARCH=mcp
export AGENTICA_WEB_SEARCH_MCP_URL=https://my-bing-mcp.example.com/mcp
export AGENTICA_WEB_SEARCH_MCP_TOOL=bing_search
export AGENTICA_WEB_SEARCH_API_KEY=sk-xxx          # 可选，作为 Bearer header 发送
export AGENTICA_WEB_SEARCH_MCP_COUNT_ARG=numResults # 可选，条数参数名

from agentica.tools.builtin import register_web_search_backend

register_web_search_backend(
    "bing",
    lambda api_key: MyBingTool(api_key=api_key),
    "bing_search",                # async (queries, max_results) -> str
    key_env="BING_API_KEY",
    key_required=True,
)
# 之后 provider="bing" 或 AGENTICA_WEB_SEARCH=bing 都可用

async def my_search(queries, max_results=5) -> str:
    return await my_mcp_client.call_tool("bing_search", {"q": queries, "n": max_results})

agent = Agent(tools=[BuiltinWebSearchTool(search_fn=my_search), ...])

from agentica import DeepAgent, OpenAIChat

# DeepAgent 自动包含以上所有工具
agent = DeepAgent(
    model=OpenAIChat(id="gpt-4o"),
    work_dir="./my-project",
)

from agentica.tools.baidu_search_tool import BaiduSearchTool
from agentica.tools.duckduckgo_tool import DuckDuckGoTool
from agentica.tools.search_serper_tool import SearchSerperTool  # SERPER_API_KEY
from agentica.tools.search_exa_tool import SearchExaTool        # EXA_API_KEY

agent = Agent(tools=[DuckDuckGoTool()])

from agentica.tools.url_crawler_tool import UrlCrawlerTool
from agentica.tools.jina_tool import JinaTool          # Jina Reader API
from agentica.tools.browser_tool import BrowserTool    # playwright 浏览器自动化

agent = Agent(tools=[UrlCrawlerTool()])

from agentica.tools.builtin import BuiltinExecuteTool  # Shell 命令执行
from agentica.tools.code_tool import CodeTool          # Python 代码沙箱执行

agent = Agent(tools=[BuiltinExecuteTool(timeout=30)])

from agentica.tools.arxiv_tool import ArxivTool        # Arxiv 论文搜索
from agentica.tools.wikipedia_tool import WikipediaTool
from agentica.tools.yfinance_tool import YFinanceTool  # 金融数据
from agentica.tools.weather_tool import WeatherTool
from agentica.tools.sql_tool import SqlTool            # 数据库查询

agent = Agent(tools=[
    ArxivTool(max_results=5),
    YFinanceTool(),
])

from agentica.tools.dalle_tool import DalleTool              # DALL-E 图像生成
from agentica.tools.cogview_tool import CogViewTool          # 智谱 CogView
from agentica.tools.image_analysis_tool import ImageAnalysisTool
from agentica.tools.ocr_tool import OcrTool                  # 文字识别

agent = Agent(tools=[DalleTool(), ImageAnalysisTool()])

from agentica.tools.ask_user_question_tool import AskUserQuestionTool

def my_input_handler(prompt: str, options=None) -> str:
    """自定义输入处理（如 Web UI 弹窗）"""
    print(f"Agent asks: {prompt}")
    return input("> ")

agent = Agent(
    model=ZhipuAI(),
    tools=[AskUserQuestionTool(input_callback=my_input_handler)],
)
# Agent 需要确认时会调用 ask_user_question 工具：
# "确定要删除 production 数据库吗？(yes/no)"

from agentica.tools.mcp_tool import McpTool

# stdio 模式（本地进程）
async with McpTool("python mcp_server.py") as mcp:
    agent = Agent(model=ZhipuAI(), tools=[mcp])
    result = await agent.run("用 MCP 工具做计算")

# SSE 模式（远程服务）
async with McpTool(url="http://localhost:8080/sse") as mcp:
    agent = Agent(model=ZhipuAI(), tools=[mcp])

from agentica.agent.config import ToolConfig

agent = Agent(
    model=ZhipuAI(),
    tools=[...],
    tool_config=ToolConfig(
        tool_call_limit=20,            # 单次 run 最多调用 20 次工具
    ),
)

from agentica.guardrails.tool import ToolInputGuardrail, ToolOutputGuardrail
from agentica.tools.builtin import BuiltinExecuteTool

def check_no_rm_rf(tool_name: str, args: dict) -> bool:
    """阻止危险的 rm -rf 命令"""
    if tool_name == "execute" and "rm -rf" in args.get("command", ""):
        return False  # 拦截
    return True

agent = Agent(
    model=ZhipuAI(),
    tools=[BuiltinExecuteTool()],
    tool_input_guardrails=[ToolInputGuardrail(check_no_rm_rf)],
)

# 差：描述模糊
def process(data: str) -> str:
    """处理数据"""
    ...

# 好：描述精确，包含适用场景
def parse_csv_to_json(csv_content: str, delimiter: str = ",") -> str:
    """将 CSV 格式文本转换为 JSON 数组

    适用于：数据格式转换、从 CSV 文件提取结构化数据
    Args:
        csv_content: CSV 格式的文本内容（含表头行）
        delimiter: 分隔符，默认为逗号，Excel 导出常用分号 ";"
    """
    ...

from typing import Literal

def export_report(
    format: Literal["pdf", "html", "markdown"],
    include_charts: bool = True,
) -> str:
    """导出分析报告"""
    ...
# LLM 收到 enum: ["pdf", "html", "markdown"]，不会乱传

def query_database(sql: str) -> str:
    """执行 SQL 查询"""
    try:
        result = db.execute(sql)
        return json.dumps(result, ensure_ascii=False)
    except Exception as e:
        # 返回描述性错误，让 LLM 能理解并修正
        return f"Error: {type(e).__name__}: {e}. Hint: Check column names with list_tables() first."

# 推荐：3-7 个相关工具（LLM 上下文有限）
agent = Agent(tools=[search_tool, crawl_tool, analyze_tool])

# 避免：超过 10 个工具（LLM 选择准确率下降）