Ag2 Add Custom Tool

Add a custom Python tool

When to use

The user wants their Agent to take a real-world action: hit an API, query a database, compute something, return an image. If they want shipped tools (web search, code exec, shell), see ag2-use-builtin-tools and ag2-shell-tool instead.

60-second recipe

from autogen.beta import Agent, tool
from autogen.beta.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],
)

The @tool decorator generates the LLM-facing schema from the function signature, type hints, and docstring. The docstring is the description the LLM sees — write it for an LLM reader, not just a human.

You can also pass plain undecorated functions in tools=[...] and AG2 wraps them automatically:

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

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

Or attach a tool to an existing agent with @agent.tool:

agent = Agent("calc")

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

Sync vs async

Both def and async def are supported. Synchronous tools run in a thread by default so blocking I/O does not freeze the event loop. For ultra-fast pure-Python tools, opt out:

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

Native async tools run in the main event loop directly:

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()

Validating inputs with Pydantic Field

Use Annotated[T, Field(...)] to give the LLM strict bounds. The framework forwards these into the JSON Schema:

from typing import Annotated
from pydantic import Field
from autogen.beta 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."

You can also override the tool name and description on the decorator:

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

Returning typed Input / ToolResult

A plain str return is wrapped in TextInput automatically. For richer payloads, return an Input subtype or compose with ToolResult:

from autogen.beta 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},
    )

For raw bytes of arbitrary format, use BinaryInput(data=..., media_type="application/pdf").

End the turn early with final=True

When the tool already knows the exact final answer, skip the extra LLM round-trip:

from autogen.beta 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)

A final=True ToolResult must contain exactly one part (TextInput or DataInput).

Dependency injection (Context / Inject / Variable / Depends)

Tools can pull execution-time values without exposing them to the LLM. See references/dependency_injection.md for the full table; the basics:

from typing import Annotated
from autogen.beta 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."""
    ...

Inject annotations are stripped from the LLM-facing schema — they're an internal injection mechanism.

Going deeper

• references/dependency_injection.md — Context vs Inject vs Variable vs Depends, defaults, factories, mutability, overrides.
• website/docs/beta/tools/tools.mdx — full @tool reference, including the synthesized JSON Schema.
• website/docs/beta/depends.mdx — Depends lifecycle, yield-based teardown, caching, test overrides.
• website/docs/beta/inputs/inputs.mdx — the Input factory hierarchy and provider support matrix.
• website/docs/beta/tools/toolkits.mdx — bundle related tools into a reusable Toolkit.
• website/docs/beta/tools/tool_middleware.mdx — async hooks around a single tool (validation, redaction, approval — see also ag2-hitl).

Common pitfalls

• Vague docstring — the LLM uses it to decide when to call the tool. "Calculates shipping cost based on destination and weight" is much better than "Shipping calc".
• No type hints — without them the framework can't generate a useful JSON Schema; the LLM may not call your tool at all.
• Blocking the event loop — if you write def (sync) tool with heavy CPU or network and pass sync_to_thread=False, the loop blocks. Default behaviour (run in a thread) is safe; only opt out for cheap pure-Python work.
• Function-level imports inside tools — repo convention disallows them. Hoist import to module top.
• Nested function definitions inside the tool body — also disallowed (recreates the function on every call).
• Returning `dict` directly when you wanted structured data — wrap it in DataInput(...) so the framework treats it as structured rather than coercing to text.
• Forgetting `final=True` requires exactly one part — combining multiple Inputs with final=True will raise.

Dependency injection in AG2 beta tools

Four annotations let a tool pull values from execution context without exposing them to the LLM. They use the same fast_depends machinery as FastAPI.

At a glance

Resolution annotations do not appear in the LLM-facing tool schema.

Context — direct access

from autogen.beta import Context, tool

@tool
def query(query: str, context: Context) -> str:
    db = context.dependencies["db"]
    api_key = context.variables.get("api_key")
    return f"{api_key}: {db.execute(query)}"

Context exposes .dependencies, .variables, .prompt, .stream, and .input(...) for HITL.

Inject — typed dependency lookup

from typing import Annotated
from autogen.beta import Inject, tool

@tool
def fetch(
    url: str,
    http_session: Annotated[object, Inject()],   # looks up "http_session" in deps
    db: Annotated[object, Inject("database")],   # looks up "database" in deps
) -> str:
    ...

Provide dependencies on the agent (broad) or per-ask (narrow). Per-ask wins on collision:

agent = Agent("data", dependencies={"db": prod_db, "http_session": shared_session})
await agent.ask("Query the user table", dependencies={"db": readonly_db})

Defaults if missing:

client: Annotated[object | None, Inject(default=None)]
client: Annotated[object, Inject(default_factory=DefaultClient)]

Variable — scalar state

from typing import Annotated
from autogen.beta import Variable, tool

@tool
def fetch_user(
    user_id: str,
    api_key: Annotated[str, Variable()],
    theme: Annotated[str, Variable(default="dark")],
) -> str:
    ...

Provide variables on the agent or per-ask, same merge/override rules as deps. Variables are mutable inside tools and persist across tool calls in the same conversation:

@tool
def authenticate(context: Context) -> str:
    context.variables["auth_token"] = "abc-123"
    return "Authenticated"

@tool
def fetch_secure(auth_token: Annotated[str | None, Variable(default=None)]) -> str:
    if not auth_token:
        return "Not authenticated"
    return f"Data with token {auth_token}"

Depends — computed dependencies

For something that must be evaluated at execution time (auth checks, short-lived sessions, side effects):

from typing import Annotated
from autogen.beta import Depends, tool

def verify_permissions(user_id: int) -> None:
    if not _allowed(user_id):
        raise PermissionDenied(user_id)

@tool
def delete_user(
    user_id: int,
    auth: Annotated[None, Depends(verify_permissions)],
) -> str:
    return f"User {user_id} deleted."

verify_permissions runs before the tool body. Its return value is injected as auth (here ignored — the dependency is purely for its side effect / raise).

Yield-based teardown

def get_db_session():
    print("opening session")
    session = "db_session_object"
    yield session
    print("closing session")  # runs after the tool finishes

@tool
def fetch_records(db: Annotated[str, Depends(get_db_session)]) -> str:
    return "Records fetched."

Combining Depends and Inject

Inject for the long-lived pool, Depends for the short-lived per-call resource:

def get_session(pool: Annotated[Pool, Inject("database_pool")]) -> Session:
    session = pool.acquire()
    yield session
    session.release()

@tool
def fetch(session: Annotated[Session, Depends(get_session)]) -> str:
    ...

agent = Agent("data", tools=[fetch], dependencies={"database_pool": Pool()})

Caching

If multiple parameters declare the same Depends(fn), fn is called once and cached for the rest of that tool call. Pass use_cache=False to force re-evaluation each time.

Test overrides

def get_production_db():
    raise Exception("Do not call in tests!")

@tool
def read_data(db: Annotated[object, Depends(get_production_db)]) -> str:
    return "Data"

agent = Agent("test", tools=[read_data])
agent.dependency_provider.override(get_production_db, lambda: "mock_db")

For Inject overrides, just pass dependencies={...} to agent.ask(...).

When to pick which

• Need the whole context object? → Context.
• Pre-built object, used as-is? → Inject.
• Plain scalar / config value? → Variable.
• Computed at call time, possibly with cleanup? → Depends.