ABZ Agent SDK

A lightweight Python SDK for building AI agents powered by Google Gemini and Groq. One consistent API, two providers, zero boilerplate.

pip install abz-agents

---

Table of Contents

• Quick Start
• Installation
• Providers & Models
• The Agent
• AgentResult
• Memory
• Tools
  • function_tool decorator
  • Tool base class
  • FunctionTool (manual)
  • Built-in Tools
• Multi-Step / Iterative Mode
• Dynamic Instructions
• Structured Output
• Guardrails
• Handoffs
• Interactive Mode
• Agent as a Tool
• Verbose Mode
• CLI
• Environment Variables
• Changelog

---

Quick Start

from dotenv import load_dotenv
load_dotenv()

from abzagent import Agent, Memory

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    model="gemini-2.0-flash",
    memory=Memory(),
)

result = agent.run("What is the capital of France?")
print(result.content)

Or start an interactive terminal session with no loop to write yourself — same run(), no new method to learn:

agent.run(interactive=True)

---

Installation

pip install abz-agents

Create a .env file in your project root:

GEMINI_API_KEY=your_gemini_key_here
GROQ_API_KEY=your_groq_key_here

The SDK loads .env automatically via python-dotenv. You do not need to call load_dotenv() manually, though it is harmless to do so.

---

Providers & Models

The SDK supports Google Gemini and Groq. The provider is detected automatically from the model name — no extra configuration required.

Gemini models

Model	Notes
gemini-2.0-flash	Default. Fast and capable.
gemini-2.0-flash-lite	Lightest 2.x variant
gemini-2.5-pro-exp-03-25	Highest quality (experimental)
gemini-1.5-pro	Stable high-quality
gemini-1.5-flash	Stable fast
gemini-1.5-flash-8b	Fastest / smallest

Requires GEMINI_API_KEY.

Groq models

Model	Notes
qwen/qwen3-32b	Balanced default for Groq
qwen/qwen-2.5-72b-instruct	High quality
llama-3.3-70b-versatile	Strong Llama 3.3
llama-3.1-8b-instant	Fastest Llama
mixtral-8x7b-32768	Long context
deepseek-r1-distill-llama-70b	Reasoning
gemma2-9b-it	Compact

Requires GROQ_API_KEY.

Provider auto-detection

Any model name containing qwen/, llama, mixtral, deepseek, gemma2, or gemma- is automatically routed to Groq. Everything else goes to Gemini.

# Gemini
agent = Agent(..., model="gemini-2.0-flash")

# Groq — detected automatically
agent = Agent(..., model="qwen/qwen3-32b")
agent = Agent(..., model="llama-3.3-70b-versatile")

---

The Agent

from abzagent import Agent, Memory

agent = Agent(
    name="My Agent",                        # required
    instructions="You are helpful.",        # required — string or function
    model="gemini-2.0-flash",               # optional, default: gemini-2.0-flash
    tools=[...],                            # optional list of Tool or plain functions
    handoffs=[...],                         # optional list of Handoff objects
    memory=Memory(),                        # optional, default: fresh Memory()
    verbose=False,                          # optional, print tool execution details
    max_iterations=1,                       # optional, default: 1 (single-turn)
    api_key="...",                          # optional, overrides env var
    output_type=None,                       # optional Pydantic model for structured output
    input_guardrails=[],                    # optional list of input guardrail functions
    output_guardrails=[],                   # optional list of output guardrail functions
)

result = agent.run("Hello!")
print(result.content)

agent.run(user_message, *, context=None)

Runs the agent and returns an AgentResult. The optional context argument is forwarded to dynamic instructions functions and guardrails via RunContextWrapper.

agent.register_tool(tool)

Add a tool after the agent has been created.

agent.register_tool(my_tool)

---

AgentResult

Every call to agent.run() returns an AgentResult:

Attribute	Type	Description
.content	str	The final text response
.steps	list[str]	All intermediate model outputs and tool results
.parsed	Any	Typed Pydantic object when output_type is set

result = agent.run("What is 2 + 2?")
print(result.content)   # "4"
print(result.steps)     # list of raw model/tool outputs

---

Memory

Memory is a simple conversation buffer that stores every turn and replays it as context on each call to agent.run().

from abzagent import Agent

agent = Agent(
    name="ABZ Helper",
    instructions="Be concise and use tools efficiently.",
    model="gemini-2.0-flash",
)

print(agent.run("What is 2 + 2?").content)

Memory API

memory = Memory()
memory.remember("user", "Hello")          # add a message
messages = memory.load()                   # list of Message objects
prompt_text = memory.to_prompt()           # render as chat transcript string

Each Message has .role (user, assistant, tool, system) and .content.

---

Tools

Tools extend what an agent can do. The model decides when to call a tool by emitting a single-line JSON object:

{"tool": "tool_name", "args": {"key": "value"}}

The SDK parses this, validates arguments with Pydantic, calls the tool, and feeds the result back to the model.

function_tool decorator

The fastest way to create a tool. Decorate any Python function — the SDK reads its signature, type hints, and docstring to build the schema automatically.

from abzagent import function_tool, Agent

@function_tool
def get_weather(city: str) -> str:
    """Return current weather for a city."""
    return f"Sunny, 24°C in {city}."

agent = Agent(
    name="WeatherBot",
    instructions="You help with weather questions.",
    model="gemini-2.0-flash",
    tools=[get_weather],
)

result = agent.run("What's the weather in Istanbul?")
print(result.content)

Both forms are supported:

@function_tool          # no parentheses
@function_tool()        # with parentheses — required when passing options

@function_tool(
    name_override="weather",
    description_override="Get the weather for a city.",
)
def get_weather(city: str) -> str: ...

Supported argument types: str, int, float, bool, list, dict, Optional[T], List[T], Dict[K, V], Annotated[T, Field(...)], Pydantic BaseModel, dataclass, TypedDict.

Async tools are fully supported:

import asyncio

@function_tool
async def fetch_data(url: str) -> str:
    """Fetch content from a URL."""
    await asyncio.sleep(0)   # real async work here
    return f"Data from {url}"

Rich argument descriptions via Annotated or docstring:

from typing import Annotated
from pydantic import Field

@function_tool
def search(
    query: Annotated[str, Field(description="What to search for")],
    max_results: Annotated[int, Field(description="Number of results")] = 5,
) -> str:
    """Search the web."""
    ...

---

Tool base class

For full control, subclass Tool directly:

from pydantic import BaseModel
from abzagent import Agent
from abzagent.core.tools import Tool

class SearchSchema(BaseModel):
    query: str
    max_results: int = 5

class SearchTool(Tool):
    name = "search"
    description = "Search the web for information."
    schema = SearchSchema

    def run(self, **kwargs) -> str:
        query = kwargs["query"]
        n = kwargs["max_results"]
        return f"Results for '{query}' (top {n}): ..."

agent = Agent(..., tools=[SearchTool()])

---

FunctionTool (manual)

For OpenAI-compatible JSON-schema style tool definitions:

from abzagent.core.tools import FunctionTool

tool = FunctionTool(
    name="calculator",
    description="Evaluate an arithmetic expression.",
    params_json_schema={
        "type": "object",
        "properties": {
            "expression": {"type": "string", "description": "e.g. 2 + 3 * 4"}
        },
        "required": ["expression"],
    },
    python_fn=lambda expression: str(eval(expression)),  # replace with safe impl
)

Or with a JSON handler (receives raw args as a JSON string):

import json

tool = FunctionTool(
    name="echo",
    description="Echo back the input.",
    on_invoke_tool=lambda ctx, args_json: json.loads(args_json)["text"],
)

---

Built-in Tools

The SDK ships two ready-to-use tools in abzagent.Tools.

MathTool — safe arithmetic calculator

Evaluates arithmetic expressions using a stdlib ast-based whitelist evaluator. Accepts + - * / ** % // and parentheses. Rejects all other Python — imports, function calls, attribute access, lambdas, etc.

from abzagent.Tools.tools_math import MathTool

agent = Agent(
    name="MathBot",
    instructions="You solve arithmetic problems.",
    model="gemini-2.0-flash",
    tools=[MathTool()],
)

result = agent.run("What is (1.5 + 2.5) * 2?")
print(result.content)   # 8.0

Direct use:

from abzagent.Tools.tools_math import safe_eval

print(safe_eval("2 + 3 * 4"))   # 14
print(safe_eval("(1.5 + 2.5) * 2"))   # 8.0

TimeTool — current time with timezone

from abzagent.Tools.tools_time import TimeTool

agent = Agent(
    name="ClockBot",
    instructions="You tell the time.",
    model="gemini-2.0-flash",
    tools=[TimeTool()],
)

result = agent.run("What time is it in Tokyo?")
print(result.content)

Direct use:

from abzagent.Tools.tools_time import TimeTool

tool = TimeTool()
print(tool.run(timezone="America/New_York"))   # 2026-07-29T10:23:45-04:00
print(tool.run())                              # UTC

Accepts any IANA timezone string ("Europe/London", "Asia/Tokyo", "US/Pacific", etc.).

---

Multi-Step / Iterative Mode

By default, max_iterations=1 — one model call, one optional tool call, done. Set max_iterations higher to let the agent loop: call tools, get results, keep reasoning until it produces a final answer.

agent = Agent(
    name="Researcher",
    instructions="Answer questions thoroughly. Use tools as needed.",
    model="gemini-2.0-flash",
    tools=[search_tool, calculator_tool],
    max_iterations=5,
)

result = agent.run("What is the GDP of Turkey divided by its population?")
print(result.content)
print(result.steps)   # see every model output and tool result

In each iteration the agent either:

• Emits a JSON tool call → SDK executes the tool → result fed back as next prompt
• Emits a plain text answer → loop ends and AgentResult is returned

If the iteration limit is reached without a final answer, the last model output is returned with a prefix note.

---

Dynamic Instructions

instructions can be a function instead of a string. It is called at the start of every run() and receives a RunContextWrapper and the Agent itself, letting you build instructions dynamically from context, user data, or runtime state.

from abzagent.core.agent import RunContextWrapper

def my_instructions(ctx: RunContextWrapper, agent) -> str:
    user_name = ctx.context.get("user_name", "there") if ctx.context else "there"
    return f"You are a helpful assistant. The user's name is {user_name}."

agent = Agent(
    name="PersonalBot",
    instructions=my_instructions,
    model="gemini-2.0-flash",
)

result = agent.run("Hello!", context={"user_name": "Abu"})
print(result.content)

Async instruction functions are supported:

async def async_instructions(ctx, agent) -> str:
    return "Dynamic instructions from async source."

RunContextWrapper attributes:

Attribute	Description
.current_agent	The agent running this turn
.target_agent	Same as current (used in handoff chains)
.memory	The agent's Memory instance
.steps	Steps accumulated so far in this run() call
.context	The context value passed to agent.run()

---

Structured Output

Set output_type to any Pydantic model to have the agent's response automatically parsed into that type. The parsed object is available on AgentResult.parsed.

from pydantic import BaseModel
from abzagent import Agent

class WeatherReport(BaseModel):
    city: str
    temperature_c: float
    condition: str

agent = Agent(
    name="WeatherParser",
    instructions="Extract weather info as JSON.",
    model="gemini-2.0-flash",
    output_type=WeatherReport,
)

result = agent.run("Istanbul: 24°C, sunny.")
report: WeatherReport = result.parsed
print(report.city)           # Istanbul
print(report.temperature_c)  # 24.0

---

Guardrails

Guardrails are validation functions that run on input before the agent processes it and/or on output before it is returned. If a guardrail's tripwire is triggered, an exception is raised immediately.

Input guardrail

from abzagent.core.guardrails import input_guardrail, GuardrailFunctionOutput

@input_guardrail
def no_profanity(ctx, agent, user_input: str) -> GuardrailFunctionOutput:
    bad_words = ["spam", "hack"]
    triggered = any(w in user_input.lower() for w in bad_words)
    return GuardrailFunctionOutput(
        output_info={"checked": True},
        tripwire_triggered=triggered,
        reason="Profanity detected." if triggered else None,
    )

agent = Agent(
    name="SafeBot",
    instructions="Be helpful.",
    model="gemini-2.0-flash",
    input_guardrails=[no_profanity],
)

try:
    agent.run("How do I hack a server?")
except Exception as e:
    print(e)   # Input guardrail 'no_profanity' tripwire triggered.

Output guardrail

from abzagent.core.guardrails import output_guardrail, GuardrailFunctionOutput

@output_guardrail
def length_check(ctx, agent, output) -> GuardrailFunctionOutput:
    too_long = len(str(output)) > 5000
    return GuardrailFunctionOutput(
        output_info={"length": len(str(output))},
        tripwire_triggered=too_long,
        reason="Response too long." if too_long else None,
    )

agent = Agent(
    name="ConciseBot",
    instructions="Be brief.",
    model="gemini-2.0-flash",
    output_guardrails=[length_check],
)

Both sync and async guardrail functions are supported. A guardrail must return a GuardrailFunctionOutput — anything else raises TypeError.

---

Handoffs

Handoffs let one agent transfer a conversation to another specialized agent — routing, memory, and context transfer are all automatic. Each handoff is registered as a transfer_to_<agent_name> tool on the host agent; the developer never manually manages routing.

from abzagent import Agent

research_agent = Agent(name="Research", instructions="Research topics thoroughly.")
writer_agent = Agent(name="Writer", instructions="Write clear, engaging copy.")
review_agent = Agent(name="Review", instructions="Review and polish drafts.")

planner = Agent(
    name="Planner",
    instructions="Route tasks to the correct specialist.",
    handoffs=[research_agent, writer_agent, review_agent],
)

result = planner.run("Write a short blog post about black holes.")
print(result.content)
print(result.last_agent.name)  # whichever agent actually produced the final answer

Bare Agent instances in handoffs=[...] are the common case. Use the handoff() factory when you need more control:

from abzagent import Agent, handoff
from pydantic import BaseModel

class EscalationData(BaseModel):
    reason: str

def on_handoff(ctx, data: EscalationData):
    print(f"Escalating: {data.reason}")

billing_agent = Agent(name="Billing", instructions="Handle billing questions.")

support_agent = Agent(
    name="Support",
    instructions="Handle general support. Transfer billing questions to Billing.",
    handoffs=[
        handoff(
            billing_agent,
            input_type=EscalationData,   # LLM must supply a `reason` when handing off
            on_handoff=on_handoff,        # called right before the transfer happens
        ),
    ],
)

result = support_agent.run("I need help with my invoice.")
print(result.content)

Handoff errors

• CircularHandoffError — an agent tried to hand off to one already in the chain (including itself).
• MaxHandoffDepthExceededError — a handoff chain exceeded abzagent.core.handoffs.MAX_HANDOFF_DEPTH (default 5).
• InvalidHandoffTargetError — raised immediately when handoff(...) is given something that isn't an Agent.

All three are RuntimeError subclasses — the same category as guardrail tripwires (a structural safety trip, not a garden-variety model mistake). Bad handoff arguments from the model (e.g. a missing input_type field) degrade gracefully instead of raising, the same way a regular tool's bad arguments do.

Handoff context filters

Clean up conversation history before it is passed to the receiving agent, via input_filter=:

from abzagent import handoff
from abzagent.extensions.handoffs_filter import remove_all_tools, keep_last_n_turns

# Remove tool messages from history before the target agent sees it
handoff(billing_agent, input_filter=remove_all_tools)

# Keep only the last 3 messages
handoff(billing_agent, input_filter=keep_last_n_turns(3))

Handoff prompt helper

Add the recommended handoff instructions prefix to any agent's instructions:

from abzagent.extensions.handoff_prompt import prompt_with_handoff_instructions

instructions = prompt_with_handoff_instructions(
    "You are a triage agent. Route to the right specialist."
)

agent = Agent(name="Triage", instructions=instructions, ...)

---

Interactive Mode

The SDK exposes a single execution method: run(). For a quick interactive session, skip writing the loop yourself — call run(interactive=True) instead of run("..."):

from abzagent import Agent

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    model="gemini-2.5-flash",
)

agent.run(interactive=True)

🤖 Assistant started. Type 'exit' to quit.
> hello
Assistant: Hi there! How can I help you today?
> exit
👋 Goodbye!

Interactive mode is a thin wrapper that calls run(user_input) — the same method, same single-request code path — for every message. There's no separate logic to keep in sync: memory, tools, structured output, and handoffs all work automatically since each turn is just a normal run() call. It handles reading input, printing responses, exit/quit, and Ctrl+C for you, and keeps the session going instead of crashing if a single turn errors out.

---

Agent as a Tool

Any agent can be wrapped as a tool and used inside another agent, enabling nested / hierarchical agent architectures.

from abzagent import Agent

sub_agent = Agent(
    name="Summarizer",
    instructions="You summarize long text into 3 bullet points.",
    model="gemini-2.0-flash",
)

# Wrap it as a tool
summarizer_tool = sub_agent.as_tool(
    tool_name="summarize",
    tool_description="Summarize a long piece of text into bullet points.",
)

orchestrator = Agent(
    name="Orchestrator",
    instructions="You coordinate tasks. Use the summarize tool when needed.",
    model="gemini-2.0-flash",
    tools=[summarizer_tool],
)

result = orchestrator.run("Summarize the history of the internet for me.")
print(result.content)

---

Verbose Mode

Set verbose=True to print tool execution details to stdout as the agent runs. Useful for debugging multi-step agents.

agent = Agent(
    name="DebugAgent",
    instructions="Be helpful.",
    model="gemini-2.0-flash",
    tools=[my_tool],
    verbose=True,
    max_iterations=4,
)

Output includes the tool name and kwargs for every tool call.

---

CLI

The SDK installs an abz-agents command.

abz-agents setup

Interactive wizard that scaffolds a new project:

$ abz-agents setup

Welcome To ABZ Agent SDK — Project setup

Select Model Provider [gemini/groq] (default: gemini): groq
Enter your GROQ_API_KEY (leave blank to fill later): gsk_...
Agent name [My Agent]: SupportBot
Agent instructions [Be helpful and concise.]: You handle customer support.
Model id [qwen/qwen3-32b]:
Starter file name [agent.py]:

✓ Wrote .env
✓ Created agent.py

Next steps:
  1) Ensure your .env has GROQ_API_KEY set
  2) Run:  abz-agents run agent.py

abz-agents run <file.py>

Runs any Python file as __main__:

abz-agents run agent.py
abz-agents run examples/quickstart.py

---

Environment Variables

Variable	Description
GEMINI_API_KEY	Required when using Gemini models
GROQ_API_KEY	Required when using Groq models
ABZ_MODEL	Default model override (default: models/gemini-1.5-pro)
ABZ_TEMPERATURE	Sampling temperature (default: 0.4)
ABZ_MAX_ITERS	Default max iterations (default: 4)
ABZ_VERBOSE	Enable verbose mode by default (1 = on, 0 = off)

All variables are loaded from .env automatically.

---

Changelog

See CHANGELOG.md for the full release history including the v0.3.1 security fix.