ag2-add-custom-tool

用途

ag2-add-custom-tool 用于给 AG2 的 Agent 添加由 Python 代码实现的新能力。当用户希望 Agent 采取真实世界操作（调用 API、查询数据库、执行计算、返回图片）时使用。若需要的是已内置的工具（如网页搜索、代码执行、shell），文档建议改用 ag2-use-builtin-tools 与 ag2-shell-tool。

主要能力

通过 @tool 装饰器把 Python 函数变成 Agent 可调用的工具。装饰器会根据函数签名、类型提示和 docstring 生成面向 LLM 的 schema。文档强调 docstring 就是 LLM 看到的描述，应按面向 LLM 读者来编写。
也可以直接把未装饰的普通函数传入 tools=[...]，AG2 会自动包装。
可以用 @agent.tool 把工具挂到已有 agent 上。
同时支持 def 和 async def。同步工具默认在线程中运行，避免阻塞 I/O 冻结事件循环；对极快的纯 Python 工具可以用 @tool(sync_to_thread=False) 退出该行为。原生异步工具直接在主场事件循环中运行。
支持用 Annotated[T, Field(...)] 给 LLM 设定严格边界，框架会把这些约束转发进 JSON Schema。
可以在装饰器上覆盖工具名称和描述。
支持返回类型化的 Input / ToolResult：普通 str 返回值会自动包装为 TextInput；更丰富的载荷可以返回 Input 子类型或组合 ToolResult，包括 DataInput、ImageInput。任意格式的原始字节可用 BinaryInput(data=..., media_type="application/pdf")。
当工具已经知道确切的最终答案时，可用 final=True 结束回合，跳过额外的 LLM 往返。文档说明 final=True 的 ToolResult 必须只包含一个部分（TextInput 或 DataInput）。
依赖注入：工具可以在不把这些值暴露给 LLM 的情况下获取执行期的值。文档给出 Context（通过 ctx.dependencies 访问）、Inject（如 Annotated[object, Inject("http_session")]）、Variable（如 Annotated[str, Variable()]）以及 Depends 等机制。文档说明 Inject 注解会从面向 LLM 的 schema 中被剥离。

使用方式

60 秒示例：

from ag2 import Agent, tool
from ag2.config import OpenAIConfig

@tool
def calculate_shipping_cost(destination: str, weight_kg: float) -> str:
    """Calculates shipping cost for a package to a destination."""
    return "$15.00"

agent = Agent(
    "shipping",
    prompt="Use tools when helpful.",
    config=OpenAIConfig(model="gpt-4o-mini"),
    tools=[calculate_shipping_cost],
)

普通函数自动包装：

def get_weather(location: str) -> str:
    """Returns the current weather for a given location."""
    return "Sunny, 22°C"

agent = Agent("weather", tools=[get_weather])

挂到已有 agent：

agent = Agent("calc")

@agent.tool
def multiply(a: int, b: int) -> int:
    """Multiplies two integers and returns the result."""
    return a * b

纯 Python 工具退出线程：

@tool(sync_to_thread=False)
def format_name(first: str, last: str) -> str:
    """Formats a full name."""
    return f"{last.upper()}, {first.capitalize()}"

异步工具：

import aiohttp

@tool
async def fetch(url: str) -> str:
    """Fetches a URL with aiohttp."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as r:
            return await r.text()

用 Pydantic Field 校验输入：

from typing import Annotated
from pydantic import Field
from ag2 import tool

@tool
def set_temperature(
    temp: Annotated[int, Field(description="Target temperature.", ge=10, le=30)],
    mode: Annotated[str, Field(description="Mode.", pattern="^(heat|cool|auto)$")],
) -> str:
    """Sets the thermostat."""
    return f"Set to {temp}°C in {mode} mode."

覆盖工具名与描述：

@tool(name="custom_math_tool", description="Performs advanced math.")
def math_op(a: int, b: int) -> int:
    return a + b

返回类型化 Input / ToolResult：

from ag2 import DataInput, ImageInput, TextInput, ToolResult, tool

@tool
def get_status(task_id: str) -> TextInput:
    return TextInput(f"Task {task_id} is in progress.")

@tool
def get_user_profile(user_id: str) -> DataInput:
    return DataInput({"id": user_id, "name": "Alice", "role": "admin"})

@tool
def fetch_chart(chart_id: str) -> ImageInput:
    return ImageInput(f"https://charts.example.com/{chart_id}.png")

@tool
def analyze_product(product_id: str) -> ToolResult:
    """Returns image + structured metadata in one tool call."""
    return ToolResult(
        ImageInput(f"https://cdn.example.com/products/{product_id}.jpg"),
        {"id": product_id, "name": "Widget Pro", "stock": 42},
    )

用 final=True 提前结束回合：

from ag2 import ToolResult, tool

@tool
def handoff_to_human(ticket_id: str) -> ToolResult:
    """Escalates and returns the final user-facing message verbatim."""
    return ToolResult(f"Ticket {ticket_id} was escalated.", final=True)

依赖注入：

from typing import Annotated
from ag2 import Context, Inject, Variable, tool

@tool
def query_db(query: str, ctx: Context) -> str:
    """Runs a SQL query."""
    db = ctx.dependencies["db"]
    return db.execute(query)

@tool
def fetch(url: str, http: Annotated[object, Inject("http_session")]) -> str:
    """Fetches with a shared HTTP session."""
    return http.get(url).text

@tool
def send(text: str, api_key: Annotated[str, Variable()]) -> str:
    """Sends a message via the configured channel."""
    ...

适用场景

Agent 需要基于 Python 代码执行实际操作时：调用外部 API、查询数据库、做计算、返回文本、结构化数据、图片或二进制内容；需要给 LLM 设定参数约束；需要在不向 LLM 暴露敏感值的前提下注入执行期依赖；或在工具已确定最终答案时减少一次 LLM 往返。

限制与常见陷阱

。docstring 过于模糊：LLM 依靠它判断何时调用工具，文档建议用 "Calculates shipping cost based on destination and weight" 这类描述，而不是 "Shipping calc"。
没有类型提示：没有类型提示框架无法生成有用的 JSON Schema，LLM 可能完全不会调用工具。
阻塞事件循环：如果写同步（def）工具、包含重 CPU 或网络操作，又传 sync_to_thread=False，会阻塞循环。默认行为（在线程中运行）是安全的；只有在做廉价的纯 Python 工作时才退出。
工具内部使用函数级 import：仓库约定不允许，import 应提升到模块顶部。
工具体内嵌套函数定义：同样不允许，因为会在每次调用时重新创建函数。
想要结构化数据却直接返回 dict：应包装成 DataInput(...)，框架才会按结构化处理，而不是强制转成文本。
忘记 final=True 只能包含一个部分：把多个 Input 与 final=True 组合会报错。
文档还指向更深层的参考资料：references/dependency_injection.md（Context vs Inject vs Variable vs Depends、默认值、工厂、可变性、覆盖）、website/docs/user-guide/tools/tools.mdx（完整 @tool 参考，包括合成出的 JSON Schema）、website/docs/user-guide/depends.mdx（Depends 生命周期、基于 yield 的清理、缓存、测试覆盖）、website/docs/user-guide/multimodal/inputs.mdx（Input 工厂层级与提供方支持矩阵）、website/docs/user-guide/tools/toolkits.mdx（把相关工具打包成可复用 Toolkit）、website/docs/user-guide/tools/tool_middleware.mdx（围绕单个工具的异步钩子：校验、脱敏、审批，另见 ag2-hitl）。

技能介绍

用途

主要能力

使用方式

适用场景

限制与常见陷阱