Loading...
Loading...
Add a custom Python tool to an AG2 `Agent` using the `@tool` decorator. Use when the user wants to give an Agent a new capability backed by Python code (API calls, DB queries, computations, file ops). Covers sync and async tools, parameter typing, Pydantic schema customisation, returning typed `Input` / `ToolResult` (text / data / images / binary), `final=True` early-exit, and dependency injection via `Context` / `Inject` / `Variable` / `Depends`.
npx skill4agent add ag2ai/ag2-skills ag2-add-custom-toolAgentag2-use-builtin-toolsag2-shell-toolfrom 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],
)@tooltools=[...]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.toolagent = Agent("calc")
@agent.tool
def multiply(a: int, b: int) -> int:
"""Multiplies two integers and returns the result."""
return a * bdefasync def@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()FieldAnnotated[T, 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 + bInputToolResultstrTextInputInputToolResultfrom 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},
)BinaryInput(data=..., media_type="application/pdf")final=Truefrom 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)final=TrueToolResultTextInputDataInputreferences/dependency_injection.mdfrom 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."""
...Injectreferences/dependency_injection.mdContextInjectVariableDependswebsite/docs/user-guide/tools/tools.mdx@toolwebsite/docs/user-guide/depends.mdxDependswebsite/docs/user-guide/multimodal/inputs.mdxInputwebsite/docs/user-guide/tools/toolkits.mdxToolkitwebsite/docs/user-guide/tools/tool_middleware.mdxag2-hitldefsync_to_thread=FalseimportdictDataInput(...)final=TrueInputfinal=True