Tool

from ai_query import tool, Field

@tool(description="Description shown to the AI")
async def my_tool(
    param1: str = Field(description="What this parameter does"),
    param2: int = Field(description="Another parameter", default=10)
) -> str:
    return "result"

from ai_query import Field

Field(description="...", default=None)

from ai_query import tool, Field

@tool(description="Add two numbers together")
def add(
    a: int = Field(description="First number"),
    b: int = Field(description="Second number")
) -> str:
    return str(a + b)

import aiohttp
from ai_query import tool, Field

@tool(description="Fetch a webpage")
async def fetch_url(
    url: str = Field(description="The URL to fetch")
) -> str:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

@tool(description="Search a database")
async def search(
    query: str = Field(description="Search query"),
    limit: int = Field(description="Max results", default=10),
    include_archived: bool = Field(description="Include archived", default=False)
) -> str:
    # limit and include_archived have defaults
    ...

from typing import NotRequired, TypedDict

class SearchFilters(TypedDict):
    tags: list[str]
    include_archived: NotRequired[bool]

@tool(description="Search documents")
async def search_documents(
    query: str = Field(description="Search query"),
    filters: SearchFilters = Field(description="Structured search filters")
) -> str:
    ...

def run_query(payload: dict) -> str:
    return "ok"

custom_tool = tool(
    description="Run a query with a custom schema",
    parameters={
        "type": "object",
        "properties": {
            "payload": {
                "type": "object",
                "properties": {
                    "sql": {"type": "string"},
                    "limit": {"type": "integer"},
                },
                "required": ["sql"],
            }
        },
        "required": ["payload"],
    },
    execute=run_query,
)

from ai_query import ToolExecutionContext, tool

@tool
async def work(value: str, ctx: ToolExecutionContext) -> str:
    ctx.signal.throw_if_aborted()
    await ctx.emit_progress("working", data={"value": value})
    return value

from ai_query import generate_text
from ai_query.providers import google

result = await generate_text(
    model=google("gemini-2.0-flash"),
    prompt="What is 5 + 3?",
    tools={"add": add}  # Pass as dict
)

@dataclass
class ToolCall:
    id: str           # Unique call ID
    name: str         # Tool name
    arguments: dict   # Arguments passed

@dataclass
class ToolResult:
    tool_call_id: str  # Matches ToolCall.id
    tool_name: str
    result: Any
    is_error: bool = False

from ai_query import FilePart, ImagePart, TextPart, ToolOutput

ToolOutput(content=[
    TextPart(text="Inspection complete"),
    ImagePart(image=image_bytes, media_type="image/png"),
    FilePart(data=pdf_bytes, media_type="application/pdf", filename="report.pdf"),
])

result = await generate_text(
    model=google("gemini-2.0-flash"),
    prompt="Calculate 10 * 5",
    tools={"multiply": multiply}
)

# See what tools were called
for call in result.tool_calls:
    print(f"Called: {call.name}")
    print(f"Args: {call.arguments}")

# See tool results
for tr in result.tool_results:
    print(f"Result: {tr.result}")