Agno — Agents, Tools & Structured Output¶

from agno.agent import Agent
from agno.models.anthropic import Claude          # needs: uv add anthropic
from agno.models.openai import OpenAIChat, OpenAILike

Agent(model="openai:gpt-5-mini")                   # string -> OpenAIResponses (Responses API)
Agent(model="openai-chat:gpt-5-mini")              # string -> OpenAIChat (Chat Completions API)
Agent(model=Claude(id="claude-sonnet-5"))
Agent(model=OpenAIChat(id="gpt-5-mini", base_url="http://localhost:4000/v1"))   # e.g. a LiteLLM proxy
Agent(model=OpenAILike(id="my-model", base_url="http://localhost:8000/v1", api_key="local"))

agent = Agent(
    model="openai:gpt-5-mini",
    description="You are a QA assistant.",
    instructions=["Be brief.", "Cite ticket IDs."],
    expected_output="One line.",
    markdown=True,
)

You are a QA assistant.
- Be brief.
- Cite ticket IDs.

<additional_information>
- Use markdown to format your answers.
</additional_information>

<expected_output>
One line.
</expected_output>

from agno.agent import Agent


def get_order_status(order_id: str) -> str:
    """Return the delivery status of an order.

    Args:
        order_id: Order ID, for example "ord-42".
    """
    return f"{order_id}: shipped"


agent = Agent(model="openai:gpt-5-mini", tools=[get_order_status])
# schema sent to the model:
# {"type": "object",
#  "properties": {"order_id": {"type": "string", "description": "Order ID, for example \"ord-42\"."}},
#  "required": ["order_id"]}

from agno.run import RunContext


def add_item(run_context: RunContext, item: str) -> str:
    """Add an item to the shopping list."""
    run_context.session_state.setdefault("items", []).append(item)   # per-session state
    return f"added {item}"


def customer_tier(run_context: RunContext) -> str:
    """Return the current customer tier."""
    return run_context.dependencies["tier"]


agent = Agent(
    model="openai:gpt-5-mini",
    tools=[add_item, customer_tier],
    session_state={"items": []},
    dependencies={"tier": "gold"},          # add_dependencies_to_context=True also puts it in the prompt
)

from agno.tools import tool


@tool(name="lookup_order", description="Find an order by ID.", requires_confirmation=True)
def get_order(order_id: str) -> str:
    return f"{order_id}: shipped"


get_order.name               # "lookup_order"  (a agno.tools.Function now, not a plain function)
get_order.entrypoint("x")    # "x: shipped"    (the original callable, for unit tests)

from agno.tools import Toolkit


class OrderTools(Toolkit):
    def __init__(self, base_url: str, **kwargs):
        self.base_url = base_url
        super().__init__(name="order_tools", tools=[self.get_status, self.cancel], **kwargs)

    def get_status(self, order_id: str) -> str:
        """Return the order status."""
        return f"{order_id}: shipped"

    def cancel(self, order_id: str) -> str:
        """Cancel an order."""
        return f"{order_id}: cancelled"


read_only = OrderTools("https://orders.internal", include_tools=["get_status"])
list(read_only.functions)          # ['get_status']

import asyncio

from agno.agent import Agent
from agno.tools.mcp import MCPTools


async def main():
    async with MCPTools(command="uvx mcp-server-git") as git_tools:        # stdio server
        agent = Agent(model="openai:gpt-5-mini", tools=[git_tools])
        await agent.aprint_response("Summarise the last 3 commits")

    # remote server: MCPTools(url="https://example.com/mcp", transport="streamable-http")

asyncio.run(main())

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.tools import tool


@tool(requires_confirmation=True)
def issue_refund(order_id: str, amount: float) -> str:
    """Refund an order."""
    return f"refunded {amount} for {order_id}"


agent = Agent(model="openai:gpt-5-mini", tools=[issue_refund], db=SqliteDb(db_file="tmp/agents.db"))

run = agent.run("Refund 25 for ord-7")
if run.is_paused:                                           # run.status == RunStatus.paused
    for requirement in run.active_requirements:
        if requirement.needs_confirmation:
            print(requirement.tool_execution.tool_name, requirement.tool_execution.tool_args)
            requirement.confirm()                           # or requirement.reject(note="Not allowed")
    run = agent.continue_run(run_id=run.run_id, requirements=run.requirements, session_id=run.session_id)

print(run.content)

from typing import Literal

from pydantic import BaseModel, Field

from agno.agent import Agent


class BugReport(BaseModel):
    title: str
    severity: Literal["low", "medium", "high"]
    steps: list[str] = Field(default_factory=list)


agent = Agent(model="openai:gpt-5-mini", output_schema=BugReport)
run = agent.run("Checkout page crashes after clicking Pay")

if isinstance(run.content, BugReport):            # always check — see the warning below
    print(run.content.severity, run.content.steps)

from agno.agent import Agent, RunEvent

agent = Agent(model="openai:gpt-5-mini", tools=[get_order_status])

for event in agent.run("Where is ord-42?", stream=True, stream_events=True):
    if event.event == RunEvent.run_content.value:
        print(event.content, end="", flush=True)
    elif event.event == RunEvent.tool_call_started.value:
        print(f"\n[calling {event.tool.tool_name}]")

import asyncio

from agno.agent import Agent, RunOutput


async def fetch_status(order_id: str) -> str:
    """Return the delivery status of an order."""
    await asyncio.sleep(0)                        # e.g. an httpx.AsyncClient call
    return f"{order_id}: shipped"


async def main():
    agent = Agent(model="openai:gpt-5-mini", tools=[fetch_status])
    run = await agent.arun("Where is ord-42?")

    async for item in agent.arun("And ord-43?", stream=True, stream_events=True, yield_run_output=True):
        if isinstance(item, RunOutput):
            print(item.content)

asyncio.run(main())

from agno.agent import Agent
from agno.exceptions import CheckTrigger, OutputCheckError
from agno.guardrails import PIIDetectionGuardrail, PromptInjectionGuardrail
from agno.run.agent import RunInput, RunOutput


def log_input(run_input: RunInput) -> None:
    print("input:", run_input.input_content)


def block_secrets(run_output: RunOutput) -> None:
    if "sk-" in str(run_output.content):
        raise OutputCheckError("Secret in output", check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED)


agent = Agent(
    model="openai:gpt-5-mini",
    pre_hooks=[PromptInjectionGuardrail(), PIIDetectionGuardrail(), log_input],
    post_hooks=[block_secrets],
)