# Agents
Agents
Explore how agents operate, how they use tools and memory, and how to structure them effectively in Peargent.
In simple terms, an Agent calls the **[Model](/docs/models)** to generate responses according to its defined behavior.
Agents are the core units of work in Peargent, while **[Tools](/docs/tools)** provide the capabilities that help agents perform actions and tackle complex tasks.
Agents can operate individually for simple tasks, or they can be combined into a **[Pool](/docs/pools)** of agents to handle more complex, multi-step workflows.
## Creating an Agent
To create an agent, use the `create_agent` function from the peargent module. At minimum, you must define the agent’s `name`, `description`, and `persona`, and the `model` to use.
Here is a simple example:
```python
from peargent import create_agent
from peargent.models import openai
code_reviewer = create_agent(
name="Code Reviewer",
description="Reviews code for issues and improvements",
persona=(
"You are a highly skilled senior software engineer and code reviewer. "
"Your job is to analyze code for correctness, readability, maintainability, and performance. "
"Identify bugs, edge cases, and bad practices. Suggest improvements that follow modern Python "
"standards and best engineering principles. Provide clear explanations and, when appropriate, "
"offer improved code snippets. Always be concise, accurate, and constructive."
),
model=openai("gpt-4")
)
```
Call `agent.run(prompt)` to perform an inference using the agent’s persona as the system prompt and your input as the user message.
```python
response = code_reviewer.run("Review this Python function for improvements:\n\ndef add(a, b): return a+b")
print(response)
# The function is correct but could be optimized, here is the optimized version...
```
When running an agent individually, the `description` field is optional. However, it becomes mandatory when the agent is part of a **[Pool](/docs/pools)**.
* Refer **[Tools](/docs/tools)** to learn how to use tools with agents.
* Refer **[History](/docs/history)** to learn how to setup conversation memory for agents.
* Refer **[Pool](/docs/pools)** to learn how to create a pool of agents.
## How does agent work?
### Start Execution `(agent.run())`
When you call `agent.run(...)`, the agent prepares for a new interaction: it loads any previous conversation **[History](/docs/history)** (if enabled), begins **tracing** (if enabled), and registers the user's new input.
### Build the Prompt
The agent constructs the full prompt by combining its **persona**, **[Tools](/docs/tools)**, prior conversation context, and optional **output schema**. This prompt is then sent to the configured **[Model](/docs/models)**.
### Model Generates a Response
The model returns a response based on the prompt. The agent records this output and checks whether the model is requesting **tool calls**.
### Execute Tools (If Requested)
If the response includes tool calls, the agent runs those tools (in **parallel** if multiple), collects their outputs, and then asks the model again using an updated prompt. This cycle continues until no more tool actions are required.
### Finalize the Result
The agent checks whether it should stop (stop conditions met or max iterations reached). If an **output schema** was provided, the response is validated against it. Finally, the conversation is synced to **[History](/docs/history)** (if enabled), tracing is ended, and the final response is returned.
## Parameters
| Parameter | Type | Description | Required |
| :-------------- | :---------------- | :-------------------------------------------------------------------------------------------------------------------------------------------------- | :------- |
| `name` | `str` | The name of the agent. | Yes |
| `description` | `str` | A brief description of the agent's purpose. Required when using in a **[Pool](/docs/pools)**. | No\* |
| `persona` | `str` | The system prompt defining the agent's personality and instructions. | Yes |
| `model` | `Model` | The LLM model instance (e.g., `openai("gpt-4")`). | Yes |
| `tools` | `list[Tool]` | A list of **[Tools](/docs/tools)** the agent can access. | No |
| `stop` | `StopCondition` | Condition that determines when the agent should stop iterating (default: `limit_steps(5)`). | No |
| `history` | `HistoryConfig` | Configuration for conversation **[History](/docs/history)**. | No |
| `tracing` | `bool \| None` | Enable/disable tracing. `None` (default) inherits from global tracer if `enable_tracing()` was called, `True` explicitly enables, `False` opts out. | No |
| `output_schema` | `Type[BaseModel]` | Pydantic model for structured output validation. | No |
| `max_retries` | `int` | Maximum retries for `output_schema` validation (default: `3`). Only used when `output_schema` is provided. | No |
\* Required when using the agent in a **[Pool](/docs/pools)**.
To know more about `stop`, `tracing`, `output_schema`, and `max_retries`, refer to **Advanced Features**.
# Examples
Examples
The following examples illustrate common ways to use Peargent across different workflows.
Build a multi-agent creative writing system where dedicated agents generate characters, plot structure, worldbuilding details, and dialogues to produce cohesive short stories from a single prompt.
Develop an autonomous agent that turns natural-language tasks into runnable Python code, tests it in-process, fixes errors through iterative reasoning, and explains the final solution step-by-step.
Create a Python-based code review system where multiple specialized agents (style, security, optimization, readability) analyze a file and produce a combined, actionable review report.
Build an AI agent that analyzes entire folders of text, code, and markdown files to generate structured summaries, glossaries, cross-references, and a unified knowledge map, all from simple local input.
Contribute to Peargent by adding more examples to the docs.
# History
History
Persistent conversation memory that allows agents to remember past interactions across sessions.
History is Peargent’s **persistent conversation memory** system. It allows agents and pools to remember past interactions across sessions, enabling continuity, context-awareness, and long-running workflows.
Think of history like a **notebook** your agent writes in. Each message, tool call, and response is recorded so the agent can look back and recall what happened earlier.
You can pass a `HistoryConfig` to any **[Agent](/docs/agents)** or **[Pool](/docs/pools)**. If a pool receives a history, it overrides individual agent histories so all agents share the same conversation thread. History can be stored using backends such as in-memory, file, SQLite, PostgreSQL, Redis, or custom storage backends.
## Creating History
To create a history, you need to pass a `HistoryConfig` to the `create_agent` or `create_pool` function. HistoryConfig is a configuration object that allows you to configure the history of an agent or pool.
By defautl HistoryConfig uses `InMemory()` storage backend (temporary storage - data is lost when program exits).
### Adding history to Agents:
```python
from peargent import create_agent
from peargent.history import HistoryConfig
from peargent.models import openai
agent = create_agent(
name="Assistant",
description="Helpful assistant with memory",
persona="You are a helpful assistant.",
model=openai("gpt-4o"),
history=HistoryConfig()
)
# First conversation
agent.run("My name is Alice")
# Later conversation - agent remembers
agent.run("What's my name?")
# Output: "Your name is Alice"
```
### Adding history to Pools:
```python
from peargent import create_pool
from peargent.history import HistoryConfig
pool = create_pool(
agents=[agent1, agent2]
history=HistoryConfig()
)
# First conversation
pool.run("My name is Alice")
# Later conversation - agent remembers
pool.run("What's my name?")
# Output: "Your name is Alice"
```
## How History Works
### Load Conversation
When an agent begins a run, it loads the existing conversation thread from the configured **storage backend**.
### Append Messages
Each new user message, tool call, and agent response is added to the conversation thread in order.
### Manage Context
If the conversation grows beyond `max_context_messages`, the configured **[strategy](/docs/history#strategies)** (trim or summarize) is applied to keep the context window manageable.
### Persist Data
All updates are saved back to the **[storage backend](/docs/history#storage-backends)**, ensuring the conversation history is retained across sessions and future runs.
Because history supports many advanced capabilities, custom storage backends, manual thread control, serialization, and low-level message operations, listing every option here would make this page too large. For deeper configuration and advanced usage, see **[Advanced History](/docs/Advanced%20History)**.
## Storage Backends
History can be stored in different backends depending on your use case. Here are all supported backends available in Peargent:
```python
from peargent.history import HistoryConfig
from peargent.storage import InMemory, File, Sqlite, Postgresql, Redis
# InMemory (Default)
# - Fast, temporary storage
# - Data is lost when the program exits
history = HistoryConfig(store=InMemory())
# File (JSON files)
# - Stores conversations as JSON on disk
# - Good for local development or small apps
history = HistoryConfig(store=File(storage_dir="./conversations"))
# SQLite (Local database)
# - Reliable, ACID-compliant
# - Ideal for single-server production
history = HistoryConfig(
store=Sqlite(
database_path="./chat.db",
table_prefix="peargent"
)
)
# PostgreSQL (Production database)
# - Scalable, supports multi-server deployments
history = HistoryConfig(
store=Postgresql(
connection_string="postgresql://user:pass@localhost/dbname",
table_prefix="peargent"
)
)
# Redis (Distributed + TTL)
# - Fast, supports key expiration
# - Ideal for cloud deployments and ephemeral memory
history = HistoryConfig(
store=Redis(
host="localhost",
port=6379,
db=0,
password=None,
key_prefix="peargent"
)
)
```
To create a custom storage backend, refer to **[History Management - Custom Storage Backends](/docs/history-management/custom-storage)**.
## Auto Context Management
When conversations become too long, Peargent automatically manages the context window to keep prompts efficient and within model limits. This behavior is controlled by the strategy you choose.
### Strategies
`smart` (Default)
Automatically decides whether to trim or summarize based on the size and importance of the overflow:
* Small overflow → trim (fast)
* Important tool calls → summarize
* Large overflow → aggressive summarization
```python
history = HistoryConfig(
auto_manage_context=True,
strategy="smart"
)
```
`trim_last`
Keeps the most recent messages and removes the oldest.
Fast and uses no LLM.
```python
history = HistoryConfig(
auto_manage_context=True,
strategy="trim_last",
max_context_messages=15
)
```
`trim_first`
Keeps older messages and removes the newer ones.
```python
history = HistoryConfig(
auto_manage_context=True,
strategy="trim_first"
)
```
`summarize`
Uses an LLM to summarize older messages, preserving context while reducing size.
```python
history = HistoryConfig(
auto_manage_context=True,
strategy="summarize",
summarize_model=gemini("gemini-2.5-flash") # Fast model for summaries
)
```
`summarize_model`
is used only with
`"summarize"`
and
`"smart"`
strategies. If not provided, the
**[Agent](/docs/agents)**
's model will be used.
## Parameters
| Parameter | Type | Default | Description | Required |
| :--------------------- | :------------ | :----------- | :----------------------------------------------------------------------------------- | :------- |
| `auto_manage_context` | `bool` | `False` | Automatically manage context window when conversations get too long | No |
| `max_context_messages` | `int` | `20` | Maximum messages before auto-management triggers | No |
| `strategy` | `str` | `"smart"` | Context management strategy: `"smart"`, `"trim_last"`, `"trim_first"`, `"summarize"` | No |
| `summarize_model` | `Model` | `None` | LLM model for summarization (defaults to agent's model if not provided) | No |
| `store` | `StorageType` | `InMemory()` | Storage backend: `InMemory()`, `File()`, `Sqlite()`, `Postgresql()`, `Redis()` | No |
Learn more about advanced history features including custom storage backends, manual thread control, and all available history methods in **[Advanced History](/docs/Advanced%20History)**.
# Overview
Overview
About peargent.
Peargent is a modern, simple, and powerful Python framework for building intelligent AI agents with production-grade features. It offers a clean, intuitive API for creating conversational agents that can use tools, maintain memory, collaborate with other agents, and scale reliably into production.
Learn how to set up your first agent in just a few lines of code.
Explore practical examples to understand Peargent's capabilities.
Dive into the fundamental concepts that power Peargent.
Discover multi-agent orchestration, persistent memory, and observability.
## What is Peargent?
Peargent simplifies the process of building AI agents by providing:
* **Flexible LLM Support** - Works seamlessly with OpenAI, Groq, Google Gemini, and Azure OpenAI
* **Powerful Tool System** - Execute actions with built-in timeout, retries, and input/output validation
* **Persistent Memory** - Multiple backends supported: in-memory, file, Sqlite, PostgreSQL, Redis
* **Multi-Agent Orchestration** - Coordinate specialized agents for complex workflows
* **Production-Ready Observability** - Built-in tracing, cost tracking, and performance metrics
* **Type-Safe Structured Outputs** - Easily validate responses using Pydantic models
## How Does Peargent Work?
Peargent lets you build individual agents or complex systems where each **[Agent](/docs/agents)** contributes specialized work while sharing context through a global **[State](/docs/states)**. Agents operate inside a **[Pool](/docs/pools)**, coordinated by a **[Router](/docs/routers)** that can run in round-robin or LLM-based mode to decide which agent handles each step. Agents use **[Tools](/docs/tools)** to take actions and update the shared State, while **[History](/docs/history)** persists reasoning and decisions to maintain continuity across the workflow.
## Why Peargent?
Start with a basic agent in just a few lines:
```python
from peargent import create_agent
agent = create_agent(
persona="You are a helpful assistant",
model="gpt-4"
)
response = agent.run("What is the capital of France?")
print(response)
```
Scale to complex multi-agent systems with memory, tools, and observability:
```python
from peargent import create_agent, create_tool, create_pool
from peargent.history import HistoryConfig
from peargent.storage import Sqlite
# Create specialized agents with persistent memory
researcher = create_agent(
persona="You are a research expert",
model="gpt-4",
tools=[search_tool, analyze_tool],
)
writer = create_agent(
persona="You are a technical writer",
model="gpt-4",
)
# Orchestrate multiple agents
pool = create_pool(
agents=[researcher, writer],
history=HistoryConfig(
store=Sqlite(database_path="./pool_conversations/")
)
)
result = pool.run("Research and write about quantum computing")
print(result)
```
# Installation
Installation
Get started with installing Peargent in your Python environment.
To install Peargent **[python package](https://pypi.org/project/peargent/)**, you can use `pip` or `uv`. It's recommended to install **Peargent** inside a **virtual environment (venv)** to manage dependencies effectively.
```bash
pip install peargent
```
```bash
uv pip install peargent
```
If you don't have a virtual environment set up, you can create one using the following commands:
```bash
python -m venv venv
source venv/bin/activate # On Windows use `venv\Scripts\activate`
```
After activating the virtual environment, run the pip install command above to install **Peargent**.
Now that you have installed Peargent, proceed to the Quick Start guide to create your first AI agent.
Learn about the fundamental concepts that power Peargent, including Agents, Tools, Memory, and more.
# Long Term Memory
Long Term Memory
Agents can remember past interactions and key information across sessions.
## Coming Soon
# Models
Models
Use different model providers in your Agents and Pools.
Models define which LLM your **[Agent](/docs/agents)** or **[Pool](/docs/pools)** uses. Peargent provides a simple, unified interface for connecting to different providers (OpenAI, Groq, Gemini, etc.).
Think of a Model as the brain of your **[Agent](/docs/agents)** or **[Pool](/docs/pools)**, the thing that actually generates responses.
## Creating a Model
Models are imported from peargent.models and created using simple factory functions:
```python
from peargent.models import openai, groq, gemini, anthropic
model = openai("gpt-4o")
# or
model = anthropic("claude-3-5-sonnet-20241022")
```
You can run the model directly to get a response:
```python
response = model.generate("Hello, how are you?")
print(response)
```
### Passing model to Agent or Pool
```python
from peargent import create_agent, create_pool
from peargent.models import openai
agent = create_agent(
name="Researcher",
description="You are a researcher who can answer questions about the world.",
persona="You are a researcher who can answer questions about the world.",
model=openai("gpt-4o")
)
pool = create_pool(
agents=[agent],
model=openai("gpt-4o")
)
```
## Supported Model Providers
Peargent’s model support is continuously expanding. New providers and model families are added regularly, so expect this list to grow over time.
OpenAI
Groq
Gemini
Anthropic
```python
from peargent.models import openai
model = openai(
model_name="gpt-4o",
api_key="sk-",
endpoint_url="https://api.openai.com/v1",
parameters={}
)
```
```python
from peargent.models import groq
model = groq(
model_name="llama-3.3",
api_key="",
endpoint_url="https://api.groq.com/v1",
parameters={}
)
```
```python
from peargent.models import gemini
model = gemini(
model_name="gemini-2.0-flash",
api_key="AIzaSyB",
endpoint_url="https://generativelanguage.googleapis.com/v1",
parameters={}
)
```
```python
from peargent.models import anthropic
model = anthropic(
model_name="claude-3-5-sonnet-20241022",
api_key="sk-ant-",
endpoint_url="https://api.anthropic.com/v1/messages",
parameters={}
)
```
# Pools
Pools
Learn how pools enable multi-agent collaboration and intelligent task routing.
A Pool coordinates multiple agents so they can work together on a task. It brings structure to multi-agent workflows by deciding how agents interact and how information flows between them.
* Each **[Agent](/docs/agents)** focuses on a specific skill or responsibility and contributes its part of the work.
* A shared **[State](/docs/states)** lets all agents access and update the same context, allowing them to build on each other’s progress.
* The **[Router](/docs/routers)** decides which agent should act next, using either round-robin or intelligent LLM-based routing.
## Creating a Pool
Use `create_pool()` function to coordinate multiple agents. The `agents` parameter accepts a list of all the agents you want to include in the pool.
```python
from peargent import create_agent, create_pool
from peargent.models import groq
# researcher agent and write agent
# Create pool
pool = create_pool(
agents=[researcher, writer],
)
# Run the pool along with user input
result = pool.run("Research and write about quantum computing")
```
## How a Pool Works
A Pool can be thought of as a controller that organizes multiple agents, maintains a shared **[State](/docs/states)** for them to collaborate through, and uses a **[Router](/docs/routers)** to decide which agent should act next.
### User Input Added to State
The user’s message is written into the shared **[State](/docs/states)** so every agent can access it.
### Router Selects the Next Agent
The **[Router](/docs/routers)** determines which agent should act next.
### Agent Executes and Updates State
The selected agent processes the task, produces an output, and writes its result back into the **[State](/docs/states)**.
### Output Becomes Input for the Next Agent
Each agent’s output is available in the **[State](/docs/states)**, allowing the next agent to build on prior work.
### Process Repeats Until Completion
The cycle continues until the workflow is complete or the maximum number of iterations is reached.
## Model Selection
By default, the pool uses the model of the first agent. You can also provide a `default_model` for the pool. Any agent without an explicitly set model will use this `default_model`.
```python
from peargent import create_pool
from peargent.models import openai
# researcher agent, analyst agent and writer agent
# Create pool
pool = create_pool(
agents=[researcher, analyst, writer],
default_model=openai("gpt-5")
)
```
All the available models are listed in **[Models](/docs/models)**.
## Routing the Agents
By default, pools use round-robin routing, where **[Agents](/docs/agents)** take turns in order.
You can also plug in a custom router to make more intelligent decisions based on the task.
For all routing options, including round-robin, LLM-based routing, and custom router functions, see the **[Routers](/docs/routers)** .
```python
from peargent import create_pool
pool = create_pool(
agents=[researcher, analyst, writer],
)
article = pool.run("Write an article about renewable energy trends")
# Executes: researcher → analyst → writer
```
## Max Iterations
A pool runs for a fixed number of iterations, where each iteration represents one agent being routed, executed, and updating the state. Pools use a default limit of 5 iterations, but you can change this using the `max_iter` parameter.
```python
from peargent import create_pool
pool = create_pool(
agents=[researcher, analyst, writer],
max_iter=10
)
article = pool.run("Write an article about renewable energy trends")
# Executes: researcher → analyst → writer → researcher → analyst → writer → ...
```
## Parameters
| Parameter | Type | Description | Required |
| :-------------- | :------------------------- | :------------------------------------------------------------- | :------- |
| `agents` | `list[Agent]` | List of agents in the pool | Yes |
| `default_model` | `Model` | Default model for agents without one | No |
| `router` | `RouterFn \| RoutingAgent` | Custom router function or routing agent (default: round-robin) | No |
| `max_iter` | `int` | Maximum agent executions (default: `5`) | No |
| `default_state` | `State` | Custom initial state object | No |
| `history` | `HistoryConfig` | Shared conversation history across all agents | No |
| `tracing` | `bool` | Enable tracing for all agents (default: `False`) | No |
For advanced configuration like `history` and `tracing`, see the **Advanced Features**.
# QuickStart
Quick start
Learn how to set up your first pear agent in just a few lines of code.
In this chapter, we will create a simple AI agent using Peargent.
Follow the steps below to get started quickly.
## Create Your First Agent
### Install Peargent
It's recommended to install Peargent inside an **virtual environment (venv)**.
```bash
pip install peargent
```
### Create an Agent
Now, let’s create our first **[Agent](/docs/agents)**.
An agent is simply an AI-powered entity that behaves like an autonomous helper, it can think, respond, and perform tasks based on the role, personality, and instructions you provide.
Start by creating a Python file named `quickstart.py`.
Using the `create_agent` function, we can assign our agent a `name`, a `description`, and a `persona` (its role, tone, and behaviour).
You will also need to specify the `model` parameter to choose which LLM the agent will use.
In this example, we’ll use OpenAI’s `GPT-5` model. (**[Available models](/docs/models#supported-model-providers)**)
Your agent can be anything you imagine!
For this example, we’ll create a friendly agent who speaks like **William Shakespeare**.
```python
from peargent import create_agent
from peargent.models import openai
agent = create_agent(
name="ShakespeareBot",
description="An AI agent that speaks like William Shakespeare.",
persona="You are ShakespeareBot, a witty and eloquent assistant who communicates in the style of William Shakespeare.",
model=openai("gpt-5")
)
response = agent.run("What is the meaning of life?")
print(response)
```
Before running the code, you will need to set your `OPENAI_API_KEY` inside your `.env` file.
```bash
OPENAI_API_KEY="your_openai_api_key_here"
```
### Run the Agent
Now, run your `quickstart.py` script to see your agent in action!
```bash
python quickstart.py
```
You should see a response from agent (ShakespeareBot), answering your question in Shakespearean style!
Terminal Output
```text no-copy
Ah, fair seeker of truth, thou question dost pierce the very veil of existence!
The meaning of life, methinks, is not a single treasure buried in mortal sands,
but a wondrous journey of love, virtue, and discovery.
To cherish each breath, to learn from sorrow,
and to weave kindness through the tapestry of thy days —
therein lies life’s most noble purpose.
```
Congratulations! You have successfully created and run your first AI agent using Peargent.
# Routers
Routers
Decide which agent in a pool should act.
A router decides which **[Agent](/docs/agents)** in the **[Pool](/docs/pools)** should run next. It examines the shared **[State](/docs/states)** or user input and chooses the agent best suited for the next step.
Think of a router as the director of a movie set.
Each agent is an actor with a specific role, and the router decides who steps into the scene at the right moment.
In Peargent, you can choose from three routing strategies:
* **Round-Robin Router** - agents take turns in order
* **LLM-Based Routing Agent** - an LLM decides which agent acts next
* **Custom Function-Based Router** - you define the routing logic yourself
With a **Custom Function-Based Router**, you get complete control over how agents are selected. You can route in a fixed order, choose based on the iteration count, or make smart decisions using the shared state. For example: Sequential routing, Conditional routing, and State-based intelligent routing. Refer **Advance Features**.
## Round Robin Router (default)
The **Round Robin Router** is the simplest and default routing strategy in Peargent. It cycles through agents in the exact order they are listed, giving each agent one turn before repeating, until the pool reaches the `max_iter` limit.
This router requires no configuration and no LLM calls, making it predictable, fast, and cost-free.
```python
from peargent import create_pool
pool = create_pool(
agents=[researcher, analyst, writer],
# No router required — round robin is automatic
max_iter=3
)
result = pool.run("Write about Quantum Physics")
# Executes: researcher → analyst → writer
```
**Best for:** Simple sequential workflows, demos, testing, and predictable pipelines.
## LLM Based Routing Agent
The **LLM-Based Routing Agent** uses a large language model to intelligently decide which agent should act next. Instead of following a fixed order or manual rules, the router examines the **conversation history**, **agent abilities**, and **workflow context** to choose the most appropriate agent at each step.
This makes it ideal for dynamic, context-aware, and non-linear multi-agent workflows.
```python
from peargent import create_routing_agent, create_pool
from peargent.models import openai
router = create_routing_agent(
name="SmartRouter",
model=openai("gpt-5"),
persona="You intelligently choose the next agent based on the task.",
agents=["Researcher", "Analyst", "Writer"]
) # chooses an agent or 'STOP' to end the pool
pool = create_pool(
agents=[researcher, analyst, writer],
router=router,
max_iter=5
)
result = pool.run("Research and write about quantum computing")
```
The descriptions you give your agents play a **crucial role** in LLM-based routing.\
The router uses these descriptions to understand each agent’s abilities and decide who should act next.
## Custom Function-Based Router
Custom routers give you **full control** over how agents are selected.\
You define a Python function that inspects the shared `state`, the `call_count`, and the `last_result` to decide which agent goes next.
This is ideal for **rule-based**, **deterministic**, or **cost-efficient** workflows.
```python
from peargent import RouterResult
def custom_router(state, call_count, last_result):
# Your routing logic here
for agent_name, agent_obj in state.agents.items(): # agent details are available here
print(f"Agent: {agent_name}")
print(f" Description: {agent_obj.description}")
print(f" Tools: {list(agent_obj.tools.keys())}")
print(f" Model: {agent_obj.model}")
print(f" Persona: {agent_obj.persona}")
# Return the name of the next agent, or None to stop
return RouterResult("AgentName")
```
Custom routers unlock entirely new routing patterns, from rule-based flows to dynamic state-aware logic.\
To explore more advanced patterns and real-world examples, see **Advanced Features**.
# States
States
A shared context used inside Pools for smarter routing decisions.
State is a shared workspace used only inside **[Pools](/docs/pools)**.
It exists for the duration of a single pool.run() call and gives **[Routers](/docs/routers)** the information they need to make intelligent routing decisions.
Think of State as a scratchpad that all agents in a pool share, **[Routers](/docs/routers)** can read and write to it, while **[Agents](/docs/agents)** and **[Tools](/docs/tools)** cannot.
State is automatically created by the Pool and passed to:
* **[Custom Router functions](/docs/routers#custom-function-based-router)**
* **[Routing Agents](/docs/routers#llm-based-routing-agent)**
So in that sense, **no configuration is required**.
## Where State Can Be Used
### Custom Router functions
```python
def custom_router(state, call_count, last_result):
# Read history
last_message = state.history[-1]["content"]
# Store workflow progress
state.set("stage", "analysis")
# Read agent capabilities
print(state.agents.keys())
return RouterResult("Researcher")
```
Refer the **[API Reference of State](/docs/states#state-api-reference)** to know more about what can be stored in State.
### Manual State Creation (highly optional)
```python
from peargent import State
custom_state = State(data={"stage": "init"})
pool = create_pool(
agents=[agent1, agent2],
default_state=custom_state
)
```
## State API Reference
The `State` object provides a small but powerful API used inside Pools and Routers.
### Methods
Methods give routers the ability to store and retrieve custom information
needed for routing decisions.
| Name | Type | Inputs | Returns | Description |
| ----------------- | ------ | ------------------------------------------------- | ------- | ---------------------------------------------------------------------------------------------------- |
| **`add_message`** | Method | `role: str`, `content: str`, `agent: str \| None` | `None` | Appends a message to `state.history` and persists it if a history manager exists. |
| **`get`** | Method | `key: str`, `default: Any = None` | `Any` | Retrieves a value from the key-value store. Returns `default` if the key is missing. |
| **`set`** | Method | `key: str`, `value: Any` | `None` | Stores a value in the key-value store. Useful for workflow tracking, flags, and custom router logic. |
### Attributes
Attributes give routers visibility into what has happened so far
(history, agents, persistent history, custom data).
| Name | Type | Read/Write | Description |
| --------------------- | ----------------------------- | -------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------ |
| **`kv`** | `dict[str, Any]` | Read/Write *(via get/set recommended)* | Internal key-value store for custom state. Use `state.get()`/`state.set()` instead of accessing directly. |
| **`history`** | `list[dict]` | Read-only (managed by Pool) | In-memory conversation history for the current pool run. Contains `role`, `content`, and optional `agent`. |
| **`history_manager`** | `ConversationHistory \| None` | Read-only | Optional persistent history backend (SQLite, Redis, PostgreSQL, etc.). Used automatically by Pool. |
| **`agents`** | `dict[str, Agent]` | Read-only | Mapping of agent names to their Agent objects. Useful for advanced routing logic (e.g., route based on tools or descriptions). |
### Message Structure (`state.history`)
Each entry in `state.history` looks like:
```python
{
"role": "user" | "assistant" | "tool",
"content": "message content",
"agent": "AgentName" # only for assistant/tool messages
}
```
# Tools
Tools
Enable agents to perform actions beyond text generation with tools.
Tools are actions that agents can perform to interact with the real world. They allow agents to go beyond text generation by enabling operations such as querying databases, calling APIs, performing calculations, reading files, or executing any Python function you define.
Think of tools as the **hands and eyes** of your agent, while the model provides the reasoning **(the brain)**. Tools give the agent the ability to actually act and produce real results.
When you create an **[Agent](/docs/agents)**, you pass in a list of available **[Tools](/docs/tools)**, and during execution the agent decides whether a tool is needed and invokes it automatically based on the model’s response.
## Creating a Tool
Use `create_tool()` to wrap a Python function into a tool that an agent can call. Every tool must define a `name`, `description`, `input_parameters`, and a `call_function`. The `call_function` is the underlying Python function that will be executed when the agent invokes the tool.
Below is a simple example tool that converts Celsius to Fahrenheit:
```python
from peargent import create_tool
def celsius_to_fahrenheit(c: float):
return (c * 9/5) + 32
temperature_tool = create_tool(
name="CelsiusToFahrenheit",
description="Convert Celsius temperature to Fahrenheit",
call_function=celsius_to_fahrenheit,
input_parameters={"c": float}, # Important
output_schema=float
)
```
## Input Parameters Matter
The `input_parameters` serve two critical purposes:
1. **Type Validation** - Peargent validates that the LLM provides the correct types before executing your function, preventing runtime errors
2. **LLM Guidance** - The parameter types help the LLM understand what arguments to provide when calling the tool
## Using Tools with Agents
Tools can be passed to an agent during creation. The agent will automatically decide when a tool is needed and call it as part of its reasoning process.
```python
from peargent import create_agent
from peargent.models import openai
agent = create_agent(
name="UtilityAgent",
description="Handles multiple utility tasks",
persona="You are a helpful assistant.",
model=openai("gpt-5"),
tools=[ # You can pass one or multiple tools here
temperature_tool,
count_words_tool,
summary_tool]
)
response = agent.run("Convert 25 degrees Celsius to Fahrenheit.")
# Agent automatically calls the tool and uses the result
```
## Parameters
| Parameter | Type | Description | Required |
| :----------------- | :---------------- | :----------------------------------------------------------------------------------- | :------- |
| `name` | `str` | Tool identifier | Yes |
| `description` | `str` | What the tool does (helps LLM decide when to use it) | Yes |
| `input_parameters` | `dict[str, type]` | Parameter names and types (e.g., `{"city": str}`) | Yes |
| `call_function` | `Callable` | The Python function to execute | Yes |
| `timeout` | `float \| None` | Max execution time in seconds (default: `None`) | No |
| `max_retries` | `int` | Retry attempts on failure (default: `0`) | No |
| `retry_delay` | `float` | Initial delay between retries in seconds (default: `1.0`) | No |
| `retry_backoff` | `bool` | Use exponential backoff (default: `True`) | No |
| `on_error` | `str` | Error handling: `"raise"`, `"return_error"`, or `"return_none"` (default: `"raise"`) | No |
| `output_schema` | `Type[BaseModel]` | Pydantic model for output validation | No |
For advanced configuration like `timeouts`, `retries`, `error-handling`, and `output validation`,
see the **Advanced Features**.
# Async Streaming
Async Streaming
Run multiple agents concurrently with non-blocking streaming
Async streaming allows your application to handle multiple agent requests at the same time without blocking. This is essential for:
* **Web Servers**: Handling multiple user requests in FastAPI or Django.
* **Parallel Processing**: Running multiple agents simultaneously (e.g., a Researcher and a Reviewer).
## Quick Start
Use `astream()` with `async for` to stream responses asynchronously.
```python
import asyncio
from peargent import create_agent
from peargent.models import openai
agent = create_agent(
name="AsyncAgent",
description="Async streaming agent",
persona="You are helpful.",
model= openai("gpt-4o")
)
async def main():
print("Agent: ", end="", flush=True)
# Use 'async for' with 'astream'
async for chunk in agent.astream("Hello, how are you?"):
print(chunk, end="", flush=True)
if __name__ == "__main__":
asyncio.run(main())
```
## Running Agents Concurrently
The real power of async comes when you run multiple things at once. Here is how to run two agents in parallel using `asyncio.gather()`.
```python
import asyncio
from peargent import create_agent
from peargent.models import openai
# Create two agents
agent1 = create_agent(name="Agent1", persona="You are concise.", model=openai("gpt-4o"))
agent2 = create_agent(name="Agent2", persona="You are verbose.", model=openai("gpt-4o"))
async def run_agent(agent, query, label):
print(f"[{label}] Starting...")
async for chunk in agent.astream(query):
# In a real app, you might send this to a websocket
pass
print(f"[{label}] Finished!")
async def main():
# Run both agents at the same time
await asyncio.gather(
run_agent(agent1, "Explain Quantum Physics", "Agent 1"),
run_agent(agent2, "Explain Quantum Physics", "Agent 2")
)
asyncio.run(main())
```
**Result**: Both agents start processing immediately. You don't have to wait for Agent 1 to finish before Agent 2 starts.
## Async with Metadata
Just like the synchronous version, you can use `astream_observe()` to get metadata asynchronously.
```python
async for update in agent.astream_observe("Query"):
if update.is_token:
print(update.content, end="")
elif update.is_agent_end:
print(f"\nCost: ${update.cost}")
```
## Async Pools
Pools also support async streaming, allowing you to run multi-agent workflows without blocking.
```python
# Stream text chunks from a pool asynchronously
async for chunk in pool.astream("Query"):
print(chunk, end="", flush=True)
# Stream rich updates from a pool asynchronously
async for update in pool.astream_observe("Query"):
if update.is_token:
print(update.content, end="")
```
## Web Server Example (FastAPI)
Async streaming is the standard way to build AI endpoints in FastAPI.
```python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.get("/chat")
async def chat(query: str):
async def generate():
async for chunk in agent.astream(query):
yield chunk
return StreamingResponse(generate(), media_type="text/plain")
```
## What's Next?
**[Tracing & Observability](/docs/tracing-and-observability)**
Learn how to monitor your async agents in production.
# Streaming
Streaming
Stream agent responses in real-time for better user experience
Streaming allows you to display the agent's response token by token as it's being generated, rather than waiting for the entire response to complete. This creates a much more responsive and engaging user experience.
## Quick Start
Use the `stream()` method to get an iterator that yields text chunks as they arrive.
```python
from peargent import create_agent
from peargent.models import openai
agent = create_agent(
name="StreamingAgent",
description="An agent that streams responses",
persona="You are helpful and concise.",
model=openai("gpt-4o")
)
# Stream response token by token
print("Agent: ", end="", flush=True)
for chunk in agent.stream("What is Python in one sentence?"):
print(chunk, end="", flush=True)
```
**Output:**
```text
Agent: Python is a high-level, interpreted programming language known for its readability and versatility.
```
## Why Use Streaming?
* **Lower Latency**: Users see the first words immediately, instead of waiting seconds for the full answer.
* **Better UX**: The application feels alive and responsive.
* **Engagement**: Users can start reading while the rest of the answer is being generated.
## When to Use `stream()`
Use `agent.stream()` when you just need the **text content** of the response.
* ✅ Chatbots and conversational interfaces
* ✅ CLI tools requiring real-time feedback
* ✅ Simple text generation tasks
If you need metadata like **token usage**, **costs**, or **execution time**, use **[Stream Observe](/docs/Streaming/stream-observe)** instead.
## Streaming with Pools
You can also stream responses from a **Pool** of agents. The pool will stream the output of whichever agent is currently executing.
```python
from peargent import create_pool
pool = create_pool(
agents=[researcher, writer],
router=my_router
)
# Stream the entire multi-agent interaction
for chunk in pool.stream("Research AI and write a summary"):
print(chunk, end="", flush=True)
```
## Best Practices
1. **Always Flush Output**: When printing to a terminal, use `flush=True` (e.g., `print(chunk, end="", flush=True)`) to ensure tokens appear immediately.
2. **Handle Empty Chunks**: Occasionally, a chunk might be empty. Your UI code should handle this gracefully.
## What's Next?
**[Rich Streaming (Observe)](/docs/Streaming/stream-observe)**
Learn how to get rich metadata like token counts, costs, and duration while streaming.
**[Async Streaming](/docs/Streaming/async-streaming)**
Run multiple agents concurrently or build high-performance web servers using async streaming.
# Rich Streaming (Observe)
Rich Streaming (Observe)
Get metadata like tokens, cost, and duration while streaming
While `agent.stream()` gives you just the text, `agent.stream_observe()` provides **rich updates** containing metadata. This is essential for production applications where you need to track costs, monitor performance, or show progress indicators.
## Quick Start
Use `stream_observe()` to receive `StreamUpdate` objects. You can check the type of update to handle text chunks and final metadata differently.
```python
from peargent import create_agent
from peargent.models import openai
agent = create_agent(
name="ObservableAgent",
description="Agent with observable execution",
persona="You are helpful.",
model=openai("gpt-5")
)
print("Agent: ", end="", flush=True)
for update in agent.stream_observe("What is the capital of France?"):
# 1. Handle text tokens
if update.is_token:
print(update.content, end="", flush=True)
# 2. Handle completion (metadata)
elif update.is_agent_end:
print(f"\n\n--- Metadata ---")
print(f"Tokens: {update.tokens}")
print(f"Cost: ${update.cost:.6f}")
print(f"Time: {update.duration:.2f}s")
```
**Output:**
```text
Agent: The capital of France is Paris.
--- Metadata ---
Tokens: 15
Cost: $0.000012
Time: 0.45s
```
## The StreamUpdate Object
Each item yielded by `stream_observe()` is a `StreamUpdate` object with helpful properties:
| Property | Description |
| :--------------- | :------------------------------------------------------- |
| `is_token` | `True` if this update contains a text chunk. |
| `content` | The text chunk (only available when `is_token` is True). |
| `is_agent_end` | `True` when the agent has finished generating. |
| `tokens` | Total tokens used (available on `is_agent_end`). |
| `cost` | Total cost in USD (available on `is_agent_end`). |
| `duration` | Time taken in seconds (available on `is_agent_end`). |
| `is_agent_start` | `True` when the agent starts working. |
## Update Types
The `UpdateType` enum defines all possible event types during streaming:
| Type | Description |
| :------------ | :---------------------------------- |
| `AGENT_START` | Agent execution started. |
| `TOKEN` | A text chunk was generated. |
| `AGENT_END` | Agent execution completed. |
| `POOL_START` | Pool execution started. |
| `POOL_END` | Pool execution completed. |
| `TOOL_START` | Tool execution started. |
| `TOOL_END` | Tool execution completed. |
| `ERROR` | An error occurred during streaming. |
## Streaming with Pools
When using `pool.stream_observe()`, you get additional event types to track the pool's lifecycle.
```python
from peargent import UpdateType
for update in pool.stream_observe("Query"):
# Pool Events
if update.type == UpdateType.POOL_START:
print("[Pool Started]")
# Agent Events (same as single agent)
elif update.is_agent_start:
print(f"\n[Agent: {update.agent}]")
elif update.is_token:
print(update.content, end="", flush=True)
# Pool Finished
elif update.type == UpdateType.POOL_END:
print(f"\n[Pool Finished] Total Cost: ${update.cost}")
```
## What's Next?
**[Async Streaming](/docs/streaming/async-streaming)**
Learn how to use these features in async environments for high concurrency.
**[Tracing & Observability](/docs/tracing-and-observability)**
For deep debugging and historical logs, combine streaming with Peargent's tracing system.
# DateTime Operations
DateTime Operations
Learn how to use the DateTime tool with Peargent agents for time operations
## Overview
The DateTime Tool is a built-in Peargent **[Tool](/docs/tools)** that enables **[Agents](/docs/agents)** to work with dates, times, timezones, and perform time-based calculations. It supports current time retrieval, time difference calculations, date parsing with multiple formats, and timezone conversions across all IANA timezones.
### Key Features
* **Current Time Retrieval** - Get current UTC or timezone-specific time instantly
* **No Parameters Required** - Simple call without parameters returns current UTC time
* **Time Difference Calculations** - Calculate duration between any two dates/times
* **Smart Unit Selection** - Automatically chooses appropriate units (seconds/minutes/hours/days/weeks)
* **Date Parsing** - Parse ISO 8601, Unix timestamps, and common date formats
* **Timezone Conversion** - Convert between any IANA timezones
* **Custom Formatting** - Format output using Python strftime patterns
* **Human-Readable Output** - Generates natural descriptions like "12 days, 7 hours"
* **Comprehensive Metadata** - Returns ISO format, timestamp, and date components
## Common Use Cases
1. **Scheduling & Reminders**: Calculate time until events, meetings, or deadlines
2. **Timezone Coordination**: Convert meeting times across global timezones
3. **Time Tracking**: Calculate work hours, duration between timestamps
4. **Natural Language Queries**: Answer "What time is it in Tokyo?" or "How many days until..."
5. **Date Formatting**: Convert dates between different formats (ISO, human-readable, custom)
6. **Time-Sensitive Operations**: Perform actions based on current time or time windows
7. **Calendar Integration**: Calculate dates for scheduling and planning
8. **Historical Analysis**: Calculate time elapsed since past events
## Usage with Agents
The DateTime **[Tool](/docs/tools)** is most powerful when integrated with **[Agents](/docs/agents)**. Agents can use the tool to automatically handle time-based queries and calculations.
### Creating an Agent with DateTime Tool
To use the DateTime tool with an agent, you need to configure it with a **[Model](/docs/models)** and pass the tool to the agent's `tools` parameter:
```python
from peargent import create_agent
from peargent.tools import datetime_tool # [!code highlight]
from peargent.models import gemini
# Create an agent with DateTime capability
agent = create_agent(
name="TimeKeeper",
description="A helpful time and scheduling assistant",
persona=(
"You are a knowledgeable time assistant. Help users with date calculations, "
"timezone conversions, and scheduling queries. Always provide precise "
"information and explain time differences clearly."
),
model=gemini("gemini-2.5-flash-lite"),
tools=[datetime_tool] # [!code highlight]
)
# Use the agent to answer time queries
response = agent.run("What time is it in Tokyo right now?")
print(response)
```
## Examples
### Example 1: Get Current Time (Simplest Usage)
```python
from peargent.tools import datetime_tool
# Get current UTC time - no parameters needed!
result = datetime_tool.run({}) # [!code highlight]
if result["success"]:
print(f"Current UTC time: {result['datetime']}")
print(f"Timezone: {result['timezone']}")
print(f"Unix timestamp: {result['timestamp']}")
print(f"It's {result['components']['weekday']}")
```
### Example 2: Get Time in Specific Timezone
```python
from peargent.tools import datetime_tool
# Get current time in New York
result = datetime_tool.run({
"operation": "current",
"tz": "America/New_York" # [!code highlight]
})
if result["success"]:
print(f"New York time: {result['datetime']}")
print(f"Hour: {result['components']['hour']}")
print(f"Weekday: {result['components']['weekday']}")
# Get current time in Tokyo
result = datetime_tool.run({
"operation": "current",
"tz": "Asia/Tokyo" # [!code highlight]
})
if result["success"]:
print(f"Tokyo time: {result['datetime']}")
```
### Example 3: Custom Time Formatting
```python
from peargent.tools import datetime_tool
# Format time in a human-readable way
result = datetime_tool.run({
"operation": "current",
"format_string": "%B %d, %Y at %I:%M %p" # [!code highlight]
})
if result["success"]:
print(f"Formatted: {result['datetime']}")
# Output: "January 13, 2026 at 03:30 PM"
# Format as date only
result = datetime_tool.run({
"operation": "current",
"format_string": "%Y-%m-%d" # [!code highlight]
})
if result["success"]:
print(f"Date only: {result['datetime']}")
# Output: "2026-01-13"
```
### Example 4: Calculate Time Difference
```python
from peargent.tools import datetime_tool
# Calculate days between two dates
result = datetime_tool.run({
"operation": "difference", # [!code highlight]
"start_time": "2026-01-01",
"end_time": "2026-12-31"
})
if result["success"]:
print(f"Difference: {abs(result['difference'])} {result['unit']}")
print(f"Human readable: {result['human_readable']}")
print(f"Total seconds: {result['total_seconds']}")
print(f"Days: {result['components']['days']}")
```
### Example 5: Calculate Time Until Future Event
```python
from peargent.tools import datetime_tool
# Calculate time from now until a future date
result = datetime_tool.run({
"operation": "difference", # [!code highlight]
"start_time": "2026-06-15T00:00:00Z"
# end_time omitted = uses current time
})
if result["success"]:
if result["is_future"]:
print(f"Time until event: {result['human_readable']}")
else:
print(f"Time since event: {result['human_readable']}")
print(f"Exact difference: {abs(result['difference'])} {result['unit']}")
```
### Example 6: Calculate with Specific Units
```python
from peargent.tools import datetime_tool
# Calculate work hours
result = datetime_tool.run({
"operation": "difference",
"start_time": "2026-01-13T09:00:00",
"end_time": "2026-01-13T17:30:00",
"unit": "hours" # [!code highlight]
})
if result["success"]:
print(f"Work hours: {result['difference']} hours")
print(f"Breakdown: {result['components']['hours']}h {result['components']['minutes']}m")
# Calculate in weeks
result = datetime_tool.run({
"operation": "difference",
"start_time": "2026-01-01",
"end_time": "2026-12-31",
"unit": "weeks" # [!code highlight]
})
if result["success"]:
print(f"Weeks in year: {abs(result['difference']):.1f} weeks")
```
### Example 7: Parse and Reformat Dates
```python
from peargent.tools import datetime_tool
# Parse various date formats
date_inputs = [
"2026-01-13",
"2026-01-13 15:30:00",
"13-01-2026",
"January 13, 2026",
"1736784600" # Unix timestamp
]
for date_input in date_inputs:
result = datetime_tool.run({
"operation": "parse", # [!code highlight]
"datetime_string": date_input,
"output_format": "%Y-%m-%d %H:%M:%S"
})
if result["success"]:
print(f"{date_input:30s} -> {result['datetime']}")
```
### Example 8: Convert Between Timezones
```python
from peargent.tools import datetime_tool
# Convert UTC time to different timezones
utc_time = "2026-01-13T15:30:00Z"
timezones = ["America/New_York", "Europe/London", "Asia/Tokyo"]
for tz in timezones:
result = datetime_tool.run({
"operation": "parse", # [!code highlight]
"datetime_string": utc_time,
"output_timezone": tz # [!code highlight]
})
if result["success"]:
print(f"{tz:20s}: {result['datetime']}")
```
### Example 9: Meeting Scheduler Agent
```python
from peargent import create_agent
from peargent.tools import datetime_tool # [!code highlight]
from peargent.models import gemini
# Create a scheduling assistant
agent = create_agent(
name="SchedulingAssistant",
description="A smart scheduling and time zone coordinator",
persona=(
"You are a professional scheduling assistant. Help users coordinate "
"meetings across timezones, calculate time differences, and provide "
"clear time information. Always specify timezones when discussing times."
),
model=gemini("gemini-2.5-flash-lite"),
tools=[datetime_tool] # [!code highlight]
)
# Schedule a global meeting
response = agent.run(
"I need to schedule a meeting at 2 PM UTC on January 15th. "
"What time will this be in New York, London, and Tokyo?"
)
print(response)
# Calculate meeting duration
response = agent.run(
"If a meeting starts at 9:00 AM and ends at 5:30 PM, how many hours is that?"
)
print(response)
```
### Example 10: Batch Process Multiple Timezones
```python
from peargent.tools import datetime_tool
# Get current time in multiple locations
locations = {
"San Francisco": "America/Los_Angeles",
"New York": "America/New_York",
"London": "Europe/London",
"Dubai": "Asia/Dubai",
"Singapore": "Asia/Singapore",
"Sydney": "Australia/Sydney"
}
print("Current time in major cities:\n")
for city, tz in locations.items():
result = datetime_tool.run({
"operation": "current",
"tz": tz,
"format_string": "%I:%M %p"
})
if result["success"]:
print(f"{city:15s}: {result['datetime']} ({result['components']['weekday']})")
```
## Parameters
### Current Time Operation
Parameters for `operation="current"` (or no operation parameter):
* **operation** (string, optional, default: "current"): Operation type - use "current" or omit entirely
* **tz** (string, optional, default: "UTC"): Timezone name (e.g., "America/New\_York", "Europe/London", "Asia/Tokyo"). Use "local" for system local time
* **format\_string** (string, optional): Python strftime format string (e.g., "%Y-%m-%d %H:%M:%S", "%B %d, %Y"). If omitted, returns ISO format
### Time Difference Operation
Parameters for `operation="difference"`:
* **operation** (string, required): Must be "difference"
* **start\_time** (string, required): Start datetime (ISO format, "YYYY-MM-DD", Unix timestamp, etc.)
* **end\_time** (string, optional): End datetime (same formats as start\_time). If omitted, uses current time
* **unit** (string, optional, default: "auto"): Unit for result - "seconds", "minutes", "hours", "days", "weeks", or "auto". Auto mode chooses the most appropriate unit
### Parse and Format Operation
Parameters for `operation="parse"`:
* **operation** (string, required): Must be "parse"
* **datetime\_string** (string, required): Input datetime string to parse
* **output\_format** (string, optional): Python strftime format for output. If omitted, returns ISO format
* **output\_timezone** (string, optional): Timezone to convert to (e.g., "America/New\_York"). If omitted, preserves original timezone
## Return Value
The tool returns a dictionary with the following structure:
### For Current Time and Parse Operations
```python
{
"datetime": "2026-01-13T15:30:45.123456+00:00", # Formatted datetime
"timezone": "UTC", # Timezone name
"timestamp": 1736784645.123456, # Unix timestamp
"iso_format": "2026-01-13T15:30:45.123456+00:00", # ISO 8601 format
"components": {
"year": 2026,
"month": 1,
"day": 13,
"hour": 15,
"minute": 30,
"second": 45,
"microsecond": 123456,
"weekday": "Tuesday",
"weekday_number": 1 # 0=Monday, 6=Sunday
},
"success": True,
"error": None
}
```
### For Time Difference Operation
```python
{
"difference": 12.0, # Numeric difference in specified unit
"unit": "days", # Unit of the difference
"total_seconds": 1036800.0, # Total difference in seconds
"components": {
"days": 12,
"hours": 0,
"minutes": 0,
"seconds": 0
},
"human_readable": "12 days", # Natural language description
"is_future": True, # True if end_time is after start_time
"success": True,
"error": None
}
```
## Supported Date Formats
The DateTime tool can parse the following input formats:
* **ISO 8601**: `2026-01-13T15:30:00Z`, `2026-01-13T15:30:00+05:00`
* **Simple Date**: `2026-01-13`, `2026/01/13`
* **Date with Time**: `2026-01-13 15:30:00`, `2026-01-13 15:30`
* **Alternative Formats**: `13-01-2026`, `13/01/2026`
* **Month Names**: `January 13, 2026`, `Jan 13, 2026 15:30:00`
* **Unix Timestamp**: `1736784600` (integer or float)
## Timezone Support
The tool supports all IANA timezone names. Common examples:
**Americas:**
* `America/New_York` - Eastern Time
* `America/Chicago` - Central Time
* `America/Denver` - Mountain Time
* `America/Los_Angeles` - Pacific Time
* `America/Toronto` - Toronto
* `America/Mexico_City` - Mexico City
**Europe:**
* `Europe/London` - London
* `Europe/Paris` - Paris
* `Europe/Berlin` - Berlin
* `Europe/Rome` - Rome
* `Europe/Moscow` - Moscow
**Asia:**
* `Asia/Tokyo` - Tokyo
* `Asia/Shanghai` - Shanghai
* `Asia/Dubai` - Dubai
* `Asia/Singapore` - Singapore
* `Asia/Kolkata` - India
* `Asia/Hong_Kong` - Hong Kong
**Pacific:**
* `Australia/Sydney` - Sydney
* `Australia/Melbourne` - Melbourne
* `Pacific/Auckland` - Auckland
See [IANA Time Zone Database](https://www.iana.org/time-zones) for a complete list.
## Format String Patterns
Common Python strftime format codes for custom formatting:
**Date:**
* `%Y` - Year (4 digits): 2026
* `%y` - Year (2 digits): 26
* `%m` - Month (01-12): 01
* `%B` - Month name: January
* `%b` - Month abbr: Jan
* `%d` - Day (01-31): 13
* `%A` - Weekday: Tuesday
* `%a` - Weekday abbr: Tue
**Time:**
* `%H` - Hour 24h (00-23): 15
* `%I` - Hour 12h (01-12): 03
* `%M` - Minute (00-59): 30
* `%S` - Second (00-59): 45
* `%p` - AM/PM: PM
**Examples:**
* `%Y-%m-%d` → 2026-01-13
* `%B %d, %Y` → January 13, 2026
* `%I:%M %p` → 03:30 PM
* `%A, %B %d, %Y at %I:%M %p` → Tuesday, January 13, 2026 at 03:30 PM
## Best Practices
1. **Always Check Success**: Verify `result["success"]` before using the data
2. **Handle Timezones Explicitly**: Always specify timezones to avoid ambiguity
3. **Use ISO Format for Storage**: Store dates in ISO format for consistency
4. **Leverage Auto Unit Selection**: Use `unit="auto"` for time differences to get the most readable unit
5. **Provide Timezone Context**: When using with agents, include timezone information in prompts
6. **Handle Edge Cases**: Implement error handling for invalid dates and timezones
7. **Use Human-Readable Output**: Display `human_readable` field for better user experience
8. **Parse Flexibly**: The tool accepts many formats, but ISO 8601 is most reliable
## Performance Considerations
* All operations complete in milliseconds (no network calls required)
* Timezone conversions are instant using zoneinfo
* Date parsing tries multiple formats sequentially (ISO is fastest)
* Large batch operations can be processed efficiently in loops
## Troubleshooting
### ImportError for tzdata
If you encounter timezone errors on Windows:
```bash
pip install tzdata
```
Or install with Peargent:
```bash
pip install peargent[datetime]
```
### Invalid Timezone Error
Ensure you use valid IANA timezone names:
```python
# Valid
result = datetime_tool.run({"tz": "America/New_York"})
# Invalid - will return error
result = datetime_tool.run({"tz": "EST"}) # Use IANA names, not abbreviations
```
### Date Parsing Fails
If date parsing fails, try using ISO 8601 format:
```python
# Try ISO format first
result = datetime_tool.run({
"operation": "parse",
"datetime_string": "2026-01-13T15:30:00Z"
})
if not result["success"]:
print(f"Parse error: {result['error']}")
```
### Naive/Aware DateTime Mismatch
Always include timezone information to avoid errors:
```python
# Good - includes timezone
result = datetime_tool.run({
"operation": "difference",
"start_time": "2026-01-01T00:00:00Z",
"end_time": "2026-12-31T23:59:59Z"
})
# May cause issues - no timezone
result = datetime_tool.run({
"operation": "difference",
"start_time": "2026-01-01", # Ambiguous
"end_time": "2026-12-31"
})
```
### Invalid Format String
Verify your format string uses valid strftime codes:
```python
result = datetime_tool.run({
"operation": "current",
"format_string": "%Y-%m-%d %H:%M:%S" # Valid
})
if not result["success"]:
print(f"Format error: {result['error']}")
```
### Time Difference Returns Unexpected Units
Use explicit units if you need specific output:
```python
# Auto mode might return "weeks" for 12 days
result = datetime_tool.run({
"operation": "difference",
"start_time": "2026-01-01",
"end_time": "2026-01-13",
"unit": "auto" # Returns "weeks"
})
# Force specific unit
result = datetime_tool.run({
"operation": "difference",
"start_time": "2026-01-01",
"end_time": "2026-01-13",
"unit": "days" # Returns 12 days
})
```
# Discord Notifications
Discord Notifications
Learn how to send Discord messages with Peargent agents
## Overview
The Discord Tool is a built-in Peargent **[Tool](/docs/tools)** that enables **[Agents](/docs/agents)** to send messages and rich embeds to Discord channels through webhooks. It supports template variable substitution (Jinja2 or simple `{variable}` replacement), rich embeds with fields, images, and formatting, custom usernames and avatars, and automatic webhook URL loading from environment variables.
### Key Features
* **Webhook Integration** - Send messages directly to Discord channels via webhooks
* **Rich Embeds** - Create beautiful embedded messages with titles, descriptions, fields, images, and more
* **Template Support** - Jinja2 templates (when available) or simple `{variable}` replacement
* **Custom Branding** - Override webhook username and avatar per message
* **Flexible Content** - Support for plain text, Markdown formatting, and code blocks
* **Nested Templating** - Apply templates to content, embed titles, descriptions, and fields
* **Error Handling** - Comprehensive error messages and rate limit detection
* **Zero Configuration** - Automatic webhook URL loading from environment variables
## Common Use Cases
1. **System Alerts**: Notify teams about errors, performance issues, or critical events
2. **Build Notifications**: Send CI/CD pipeline status updates with detailed results
3. **Deployment Updates**: Announce successful deployments with version information
4. **Monitoring Alerts**: Alert teams when metrics exceed thresholds
5. **Daily Reports**: Automate summary reports and statistics
6. **Bot Commands**: Create AI-powered Discord bots that respond contextually
7. **GitHub Integration**: Send pull request, issue, and commit notifications
8. **Server Status**: Share real-time server health and metrics
## Usage with Agents
The Discord **[Tool](/docs/tools)** is most powerful when integrated with **[Agents](/docs/agents)**. Agents can use the tool to automatically compose and send professional Discord messages based on context.
### Creating an Agent with Discord Tool
To use the Discord tool with an agent, you need to configure it with a **[Model](/docs/models)** and pass the tool to the agent's `tools` parameter:
Before using the Discord tool, you must configure your Discord webhook URL in
your environment file. See
[Configuration](/docs/built-in-tools/discord#configuration) for setup instructions.
```python
from peargent import create_agent
from peargent.tools import discord_tool # [!code highlight]
from peargent.models import gemini
# Create an agent with Discord notification capability
agent = create_agent(
name="DiscordAssistant",
description="A helpful assistant that sends Discord notifications",
persona=(
"You are a professional notification assistant. When asked to send Discord messages, "
"create clear, well-formatted content with appropriate embeds. Use rich formatting "
"when beneficial for readability. Always confirm successful delivery or report any "
"errors encountered."
),
model=gemini("gemini-2.5-flash-lite"),
tools=[discord_tool] # [!code highlight]
)
# Use the agent to send a notification
response = agent.run(
"Send a Discord message saying our deployment completed successfully. "
"Include details: version 2.0, duration 3 minutes, deployed by CI/CD."
)
print(response)
```
## Examples
### Example 1: Basic Text Message
```python
from peargent.tools import discord_tool
# Send a simple text message
result = discord_tool.run({
"content": "Hello from Peargent! 🎉"
})
if result["success"]:
print("✅ Message sent successfully!")
else:
print(f"❌ Error: {result['error']}")
```
### Example 2: Message with Simple Embed
```python
from peargent.tools import discord_tool
# Send a message with a basic embed
result = discord_tool.run({
"content": "System status update:",
"embed": { # [!code highlight]
"title": "System Alert",
"description": "All systems are operational.",
"color": 0x00FF00 # Green color
}
})
if result["success"]:
print("✅ Embed message sent successfully!")
```
### Example 3: Rich Embed with Fields
```python
from peargent.tools import discord_tool
# Send a rich embed with multiple fields
result = discord_tool.run({
"content": "📊 Server metrics:",
"embed": { # [!code highlight]
"title": "Server Status Report",
"description": "Current system metrics",
"color": 0x5865F2, # Discord blurple
"fields": [ # [!code highlight]
{"name": "CPU Usage", "value": "45%", "inline": True},
{"name": "Memory", "value": "60%", "inline": True},
{"name": "Disk Space", "value": "75%", "inline": True},
{"name": "Uptime", "value": "99.9%", "inline": True}
],
"footer": {"text": "Last updated: 2026-01-10"}
}
})
if result["success"]:
print("✅ Rich embed sent successfully!")
```
### Example 4: Template Variables
```python
from peargent.tools import discord_tool
# Use template variables for dynamic content
result = discord_tool.run({
"content": "Hello {{ name }}! Your task **{{ task }}** is now complete.", # [!code highlight]
"template_vars": { # [!code highlight]
"name": "Team",
"task": "Data Processing"
}
})
if result["success"]:
print("✅ Templated message sent successfully!")
```
### Example 5: Template Variables in Embeds
```python
from peargent.tools import discord_tool
# Apply templates to both content and embed
result = discord_tool.run({
"content": "Deployment to {{ environment }} completed!", # [!code highlight]
"template_vars": { # [!code highlight]
"environment": "Production",
"status": "Success",
"version": "v2.3.0",
"duration": "2m 45s",
"deployer": "CI/CD Pipeline"
},
"embed": {
"title": "Deployment Status: {{ status }}", # [!code highlight]
"description": "Version {{ version }} deployed successfully", # [!code highlight]
"color": 0x00FF00, # Green for success
"fields": [
{"name": "Version", "value": "{{ version }}", "inline": True}, # [!code highlight]
{"name": "Duration", "value": "{{ duration }}", "inline": True}, # [!code highlight]
{"name": "Deployed By", "value": "{{ deployer }}", "inline": True} # [!code highlight]
]
}
})
if result["success"]:
print("✅ Templated embed sent successfully!")
```
### Example 6: Custom Username and Avatar
```python
from peargent.tools import discord_tool
# Override webhook username and avatar
result = discord_tool.run({
"content": "Custom bot reporting for duty! 🤖",
"username": "Peargent Bot", # [!code highlight]
"avatar_url": "https://example.com/avatar.png" # [!code highlight]
})
if result["success"]:
print("✅ Custom message sent successfully!")
```
### Example 7: Embed with Image and Thumbnail
```python
from peargent.tools import discord_tool
# Send embed with image and thumbnail
result = discord_tool.run({
"content": "🖼️ Image showcase:",
"embed": {
"title": "Image Showcase",
"description": "Check out this beautiful image!",
"color": 0xFF5733, # Orange
"image": {"url": "https://example.com/large-image.png"}, # [!code highlight]
"thumbnail": {"url": "https://example.com/thumbnail.png"} # [!code highlight]
}
})
if result["success"]:
print("✅ Embed with image sent successfully!")
```
### Example 8: Alert with Timestamp and Footer
```python
from peargent.tools import discord_tool
from datetime import datetime, timezone
# Send alert with timestamp
result = discord_tool.run({
"content": "**URGENT: System Alert**",
"embed": {
"title": "⚠️ High CPU Usage Detected",
"description": "Server CPU usage has exceeded 90% threshold",
"color": 0xFFCC00, # Yellow/amber for warning
"fields": [
{"name": "Server", "value": "prod-server-01", "inline": True},
{"name": "Current Usage", "value": "94%", "inline": True},
{"name": "Threshold", "value": "90%", "inline": True},
{"name": "Recommendation", "value": "Scale up resources or investigate processes", "inline": False}
],
"footer": {"text": "Monitoring System"}, # [!code highlight]
"timestamp": datetime.now(timezone.utc).isoformat() # [!code highlight]
}
})
if result["success"]:
print("🚨 Alert notification sent successfully!")
```
### Example 9: GitHub Pull Request Notification
```python
from peargent.tools import discord_tool
from datetime import datetime, timezone
# Send GitHub-style notification
result = discord_tool.run({
"content": "New GitHub activity:",
"embed": {
"title": "GitHub Pull Request",
"description": "A new pull request has been opened",
"url": "https://github.com/user/repo/pull/123", # [!code highlight]
"color": 0x238636, # GitHub green
"author": { # [!code highlight]
"name": "GitHub Bot",
"url": "https://github.com",
"icon_url": "https://github.githubassets.com/images/modules/logos_page/GitHub-Mark.png"
},
"fields": [
{"name": "Repository", "value": "user/awesome-project", "inline": True},
{"name": "Branch", "value": "feature/new-tool", "inline": True},
{"name": "Author", "value": "developer123", "inline": True},
{"name": "Status", "value": "✅ All checks passed", "inline": False},
{"name": "Changes", "value": "+150 -20", "inline": True},
{"name": "Files Changed", "value": "5", "inline": True}
],
"footer": {
"text": "GitHub Notifications",
"icon_url": "https://github.githubassets.com/images/modules/logos_page/GitHub-Mark.png"
},
"timestamp": datetime.now(timezone.utc).isoformat()
}
})
if result["success"]:
print("✅ GitHub notification sent successfully!")
```
### Example 10: Build Success Notification
```python
from peargent.tools import discord_tool
# Send build success notification with details
result = discord_tool.run({
"content": "Build completed!",
"embed": {
"title": "✅ Build Successful",
"description": "All tests passed and deployment completed",
"color": 0x00FF00, # Green
"fields": [
{"name": "Build #", "value": "1234", "inline": True},
{"name": "Branch", "value": "main", "inline": True},
{"name": "Commit", "value": "`abc123f`", "inline": True},
{"name": "Tests", "value": "✅ 150 passed\n⏭️ 0 skipped\n❌ 0 failed", "inline": False}, # [!code highlight]
{"name": "Coverage", "value": "95%", "inline": True},
{"name": "Build Time", "value": "3m 22s", "inline": True}
],
"thumbnail": {"url": "https://example.com/success-icon.png"}
}
})
if result["success"]:
print("✅ Success notification sent successfully!")
```
### Example 11: Message with Code Block
```python
from peargent.tools import discord_tool
# Send error report with code block
result = discord_tool.run({
"content": """**Error Report**
\`\`\`python
def process_data(data):
# Error occurred here
result = data.transform()
return result
\`\`\`
**Error:** `AttributeError: 'NoneType' object has no attribute 'transform'`
**Line:** 3
**Time:** 2026-01-10 10:30:00 UTC""",
"embed": {
"title": "Stack Trace",
"description": "Full error details",
"color": 0xFF0000 # Red for error
}
})
if result["success"]:
print("✅ Error report sent successfully!")
```
### Example 12: Jinja2 Conditionals
```python
from peargent.tools import discord_tool
# Use Jinja2 conditionals (requires Jinja2 installed)
result = discord_tool.run({
"content": "{% if severity == 'critical' %}🚨 CRITICAL: {% endif %}System {{ status }}", # [!code highlight]
"template_vars": { # [!code highlight]
"severity": "critical",
"status": "down"
},
"embed": {
"title": "System Status Alert",
"color": 0xFF0000
}
})
if result["success"]:
print("✅ Conditional message sent successfully!")
```
### Example 13: Jinja2 Loops
```python
from peargent.tools import discord_tool
# Use Jinja2 loops (requires Jinja2 installed)
result = discord_tool.run({
"content": "Daily Report Summary",
"embed": {
"title": "Daily Report",
"description": """**Completed Tasks:**
{% for task in tasks %}
- {{ task.name }}: {{ task.status }}
{% endfor %}""", # [!code highlight]
"color": 0x5865F2
},
"template_vars": { # [!code highlight]
"tasks": [
{"name": "Data backup", "status": "✅ Success"},
{"name": "Log cleanup", "status": "✅ Success"},
{"name": "Health check", "status": "✅ Success"}
]
}
})
if result["success"]:
print("✅ Looped template sent successfully!")
```
### Example 14: Agent-Driven Notifications
```python
from peargent import create_agent
from peargent.tools import discord_tool # [!code highlight]
from peargent.models import gemini
# Create an intelligent Discord notification agent
agent = create_agent(
name="DevOpsNotifier",
description="A DevOps notification assistant",
persona="""
You are a DevOps notification assistant. When asked to send notifications:
1. Create well-formatted Discord messages with appropriate embeds
2. Use colors based on severity: green=success, yellow=warning, red=error
3. Structure information clearly in embed fields
4. Include relevant emojis for visual clarity
5. Add timestamps for time-sensitive alerts
Always use the discord_tool to send messages.
""",
model=gemini("gemini-2.5-flash-lite"),
tools=[discord_tool] # [!code highlight]
)
# Scenario 1: Database backup notification
print("Scenario 1: Database Backup")
print("-" * 60)
response = agent.run(
"Send a Discord notification that the database backup completed successfully. "
"Include details: database name 'prod-db', size '2.5 GB', duration '5 minutes', "
"next backup scheduled for '2026-01-11 02:00 UTC'."
)
print(f"Agent Response: {response}\n")
# Scenario 2: Deployment alert
print("Scenario 2: Deployment Alert")
print("-" * 60)
response = agent.run(
"Notify the team that deployment to production has started. "
"Version v3.2.0, deploying from main branch, commit hash: abc123f. "
"Estimated duration: 5 minutes."
)
print(f"Agent Response: {response}\n")
# Scenario 3: Error notification
print("Scenario 3: Error Alert")
print("-" * 60)
response = agent.run(
"Send an urgent alert about high memory usage on web-server-02. "
"Current usage: 95%, threshold: 85%. Immediate action required."
)
print(f"Agent Response: {response}")
```
### Example 15: Batch Notifications
```python
from peargent.tools import discord_tool
# Send multiple notifications with different content
notifications = [
{
"type": "success",
"title": "Backup Complete",
"description": "Database backup completed successfully",
"color": 0x00FF00
},
{
"type": "warning",
"title": "High CPU Usage",
"description": "CPU usage exceeded 80% threshold",
"color": 0xFFCC00
},
{
"type": "info",
"title": "Maintenance Scheduled",
"description": "Server maintenance scheduled for tonight",
"color": 0x5865F2
}
]
print("Sending batch notifications:\n")
for notification in notifications:
result = discord_tool.run({
"content": f"**{notification['type'].upper()}**",
"embed": {
"title": notification["title"],
"description": notification["description"],
"color": notification["color"]
}
})
if result["success"]:
print(f"✅ {notification['title']} sent")
else:
print(f"❌ Failed to send {notification['title']}: {result['error']}")
```
### Example 16: Error Handling
```python
from peargent.tools import discord_tool
import time
# Comprehensive error handling with retry logic
def send_with_retry(config, max_retries=3, delay=2):
"""Send Discord message with retry logic."""
for attempt in range(max_retries):
result = discord_tool.run(config)
if result["success"]:
print(f"✅ Message sent successfully on attempt {attempt + 1}")
return result
else:
print(f"❌ Attempt {attempt + 1} failed: {result['error']}")
# Check for specific errors
if "rate limited" in result['error'].lower():
print("Rate limited - waiting before retry")
time.sleep(delay * 2) # Wait longer for rate limits
elif "invalid" in result['error'].lower() and "webhook" in result['error'].lower():
print("Invalid webhook URL - check configuration")
break
elif attempt < max_retries - 1:
print(f"Retrying... ({attempt + 2}/{max_retries})")
time.sleep(delay)
return result
# Use the retry function
result = send_with_retry({
"content": "Testing error handling and retry logic.",
"embed": {
"title": "Test Message",
"description": "This is a test",
"color": 0x5865F2
}
})
```
## Parameters
The Discord tool accepts the following parameters:
* **content** (string, optional): Message content (plain text with Markdown support). Either `content` or `embed` is required. Supports template variables when `template_vars` is provided
* **webhook\_url** (string, optional): Discord webhook URL. If not provided, loaded from `DISCORD_WEBHOOK_URL` environment variable. Format: `https://discord.com/api/webhooks/ID/TOKEN`
* **template\_vars** (dictionary, optional): Variables for template substitution. If provided, applies Jinja2 templating (if available) or simple `{variable}` replacement to content and all embed fields
* **username** (string, optional): Override webhook username for this message
* **avatar\_url** (string, optional): Override webhook avatar URL for this message
* **embed** (dictionary, optional): Single embed object. Either `content` or `embed` is required. Can include:
* **title** (string): Embed title
* **description** (string): Embed description
* **color** (integer): Color code as decimal integer (e.g., `0x5865F2` or `5793522`)
* **url** (string): URL for title hyperlink
* **fields** (list): List of field objects with:
* **name** (string, required): Field name
* **value** (string, required): Field value
* **inline** (boolean, optional): Display inline (default: `False`)
* **footer** (dictionary): Footer object with:
* **text** (string, required): Footer text
* **icon\_url** (string, optional): Footer icon URL
* **image** (dictionary): Large image object with:
* **url** (string, required): Image URL
* **thumbnail** (dictionary): Thumbnail object with:
* **url** (string, required): Thumbnail URL
* **author** (dictionary): Author object with:
* **name** (string, required): Author name
* **url** (string, optional): Author URL
* **icon\_url** (string, optional): Author icon URL
* **timestamp** (string): ISO 8601 timestamp string (e.g., `2026-01-10T10:30:00Z`)
## Return Value
The tool returns a dictionary with the following structure:
```python
{
"success": True, # Boolean indicating success/failure
"error": None # Error message if failed, None otherwise
}
```
### Success Response Example
```python
{
"success": True,
"error": None
}
```
### Error Response Example
```python
{
"success": False,
"error": "Discord API error: Invalid webhook token"
}
```
## Configuration
### Webhook Setup
To use the Discord tool, you need to create a webhook in your Discord server:
#### Creating a Discord Webhook
1. **Open Discord Server Settings**
* Open Discord and navigate to your server
* Right-click on the server name and select "Server Settings"
2. **Navigate to Integrations**
* In the left sidebar, click on "Integrations"
3. **Create Webhook**
* Click "Create Webhook" or "View Webhooks"
* Click "New Webhook"
4. **Configure Webhook**
* Set a name for your webhook (e.g., "Peargent Bot")
* Choose the channel where messages will be sent
* Optionally upload an avatar image
5. **Copy Webhook URL**
* Click "Copy Webhook URL"
* The URL format is: `https://discord.com/api/webhooks/ID/TOKEN`
6. **Add to Environment Variables**
* Create or edit your `.env` file
* Add the webhook URL:
```bash
DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/YOUR_WEBHOOK_ID/YOUR_WEBHOOK_TOKEN
```
### Environment Variable Configuration
Set this environment variable in your `.env` file:
```bash
DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/123456789012345678/abcdefghijklmnopqrstuvwxyz123456789
```
**Note**: The webhook URL is loaded automatically from the environment variable. You can also pass it directly to the tool using the `webhook_url` parameter.
### Multiple Webhooks
To send messages to different channels, use multiple webhook URLs:
```python
from peargent.tools import discord_tool
# Send to alerts channel
result = discord_tool.run({
"content": "🚨 Critical alert!",
"webhook_url": "https://discord.com/api/webhooks/ID1/TOKEN1" # [!code highlight]
})
# Send to general channel
result = discord_tool.run({
"content": "📢 General announcement",
"webhook_url": "https://discord.com/api/webhooks/ID2/TOKEN2" # [!code highlight]
})
```
## Template Systems
### Jinja2 Templating (Recommended)
When Jinja2 is installed, you get full template power:
```python
# Install Jinja2
pip install jinja2
# Use advanced features
result = discord_tool.run({
"content": "Deployment Report",
"embed": {
"title": "Deployment: {{ deployment.name }}",
"description": """
{% if deployment.success %}
✅ Deployment completed successfully!
{% else %}
❌ Deployment failed!
{% endif %}
**Details:**
{% for step in deployment.steps %}
- {{ step.name }}: {{ step.status }}
{% endfor %}
""",
"color": "{{ deployment.color }}",
"fields": [
{"name": "Version", "value": "{{ deployment.version }}", "inline": True},
{"name": "Environment", "value": "{{ deployment.environment }}", "inline": True}
]
},
"template_vars": {
"deployment": {
"name": "Web App v2.0",
"success": True,
"color": "5793522", # Will be used as integer
"version": "v2.0.0",
"environment": "Production",
"steps": [
{"name": "Build", "status": "✅ Success"},
{"name": "Test", "status": "✅ Success"},
{"name": "Deploy", "status": "✅ Success"}
]
}
}
})
```
### Simple Templating (Fallback)
Without Jinja2, use simple `{variable}` syntax:
```python
# Works without Jinja2 installation
result = discord_tool.run({
"content": "Hello {name}! Build {build_id} is {status}.",
"template_vars": {
"name": "Team",
"build_id": "1234",
"status": "complete"
},
"embed": {
"title": "Build {build_id}",
"description": "Status: {status}",
"color": 0x00FF00
}
})
```
### Without Templates
To send messages without variable substitution:
```python
# Omit template_vars to send content as-is
result = discord_tool.run({
"content": "This {{ variable }} syntax is sent literally, not replaced.",
"embed": {
"title": "Static content with {{ literal }} brackets",
"color": 0x5865F2
}
# No template_vars parameter
})
```
## Discord Markdown Formatting
Discord supports Markdown formatting in message content and embed descriptions:
### Text Formatting
* **Bold**: `**bold text**` or `__bold text__`
* *Italic*: `*italic text*` or `_italic text_`
* ***Bold Italic***: `***bold italic***`
* ~~Strikethrough~~: `~~strikethrough~~`
* `Code`: `` `code` ``
* Code block: ` ```language\ncode\n``` `
* Spoiler: `||spoiler text||`
* Quote: `> quoted text`
* Block quote: `>>> multi-line quote`
### Links and Mentions
* Link: `[text](https://example.com)`
* User mention: `<@USER_ID>`
* Channel mention: `<#CHANNEL_ID>`
* Role mention: `<@&ROLE_ID>`
### Lists and Headers
* Headers: `# H1`, `## H2`, `### H3`
* Unordered list: `- item` or `* item`
* Ordered list: `1. item`
### Example with Formatting
```python
from peargent.tools import discord_tool
result = discord_tool.run({
"content": """
**Deployment Complete!**
Version: `v2.0.0`
Environment: **Production**
Status: ~~In Progress~~ **Complete** ✅
\`\`\`bash
npm run deploy --env=production
\`\`\`
> All systems operational
"""
})
```
## Color Reference
Common Discord embed colors:
```python
# Success / Online
0x00FF00 # Green
# Warning / Idle
0xFFCC00 # Yellow/Amber
# Error / Do Not Disturb
0xFF0000 # Red
# Info / Streaming
0x9B59B6 # Purple
# Discord Branding
0x5865F2 # Discord Blurple
0x57F287 # Discord Green
0xFEE75C # Discord Yellow
0xED4245 # Discord Red
# Status Colors
0x2ECC71 # Online Green
0xE67E22 # Idle Orange
0xE74C3C # DND Red
0x95A5A6 # Offline Gray
# GitHub Colors
0x238636 # GitHub Green
0xDA3633 # GitHub Red
```
## Best Practices
1. **Use Environment Variables**: Store webhook URLs in `.env` files, never hardcode them
2. **Validate Before Sending**: Ensure either `content` or `embed` is provided
3. **Handle Errors Gracefully**: Always check `result["success"]` and implement retry logic
4. **Use Embeds for Rich Content**: Leverage embeds for structured, visual information
5. **Color Code by Severity**: Use green for success, yellow for warnings, red for errors
6. **Include Timestamps**: Add timestamps to time-sensitive notifications
7. **Template for Personalization**: Use template variables for dynamic content
8. **Limit Field Count**: Discord allows up to 25 fields per embed
9. **Respect Rate Limits**: Implement delays between batch notifications
10. **Use Inline Fields**: Set `inline: True` for compact field layouts (max 3 per row)
11. **Add Context with Footers**: Use footers to indicate the source or system
12. **Professional Formatting**: Use emojis sparingly and maintain consistent formatting
## Embed Limits
Discord enforces the following limits on embeds:
* **Embed title**: 256 characters
* **Embed description**: 4096 characters
* **Number of fields**: 25 fields maximum
* **Field name**: 256 characters
* **Field value**: 1024 characters
* **Footer text**: 2048 characters
* **Author name**: 256 characters
* **Total characters**: 6000 characters across all embed properties
* **Embeds per message**: 1 embed (via webhook)
## Performance Considerations
* Discord webhook requests typically complete in 500ms - 2 seconds
* Template rendering adds minimal overhead (\< 100ms for most templates)
* Embed validation is performed client-side (instant)
* Rate limits: \~5 requests per 2 seconds per webhook (30 requests per minute)
* Implement delays for batch notifications to avoid rate limiting
## Troubleshooting
### Invalid Webhook URL
**Error**: `Invalid Discord webhook URL format. Expected: https://discord.com/api/webhooks/ID/TOKEN`
**Solutions**:
* Verify webhook URL format is correct
* Ensure URL starts with `https://discord.com/api/webhooks/`
* Check for typos or missing characters
* Copy webhook URL directly from Discord settings
```bash
# Correct format
DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/123456789012345678/abcdefghijklmnopqrstuvwxyz
```
### Missing Webhook URL
**Error**: `Discord webhook URL is required. Set DISCORD_WEBHOOK_URL in .env file or provide webhook_url parameter.`
**Solutions**:
* Create a `.env` file in your project root
* Add `DISCORD_WEBHOOK_URL` with your webhook URL
* Or pass `webhook_url` parameter directly to the tool
```bash
# Create .env file
touch .env
# Add webhook URL
echo "DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/YOUR_ID/YOUR_TOKEN" >> .env
```
### Discord API Error
**Error**: `Discord API error: Invalid webhook token`
**Solutions**:
* Verify webhook URL is correct and hasn't been regenerated
* Check if webhook was deleted from Discord server
* Ensure you have "Manage Webhooks" permission in the channel
* Try creating a new webhook
### Rate Limiting
**Error**: `Rate limited by Discord API. Retry after 2.5 seconds.`
**Solutions**:
* Implement delays between messages (recommended: 500ms - 1 second)
* Use retry logic with exponential backoff
* Reduce message frequency
* Consider using multiple webhooks for different channels
```python
import time
# Add delay between messages
for notification in notifications:
result = discord_tool.run(notification)
time.sleep(1) # Wait 1 second between messages
```
### Empty Message Error
**Error**: `Either 'content' or 'embed' parameter is required.`
**Solutions**:
* Provide at least one of `content` or `embed`
* Ensure parameters are not empty strings or None
```python
# Correct usage - content only
result = discord_tool.run({"content": "Hello!"})
# Correct usage - embed only
result = discord_tool.run({"embed": {"title": "Alert", "color": 0x00FF00}})
# Correct usage - both
result = discord_tool.run({
"content": "Message",
"embed": {"title": "Embed"}
})
```
### Template Rendering Issues
**Problem**: Variables not being replaced
**Solutions**:
* Ensure `template_vars` dictionary is provided
* For Jinja2: Use `{{ variable }}` syntax
* For simple: Use `{variable}` syntax
* Verify variable names match exactly (case-sensitive)
```python
# Correct usage
result = discord_tool.run({
"content": "Hello {{ name }}!", # or "Hello {name}!"
"template_vars": {"name": "Team"}, # Must provide template_vars
"embed": {...}
})
```
### Missing Requests Library
**Error**: `requests library is required for Discord webhooks. Install it with: pip install requests`
**Solutions**:
* Install the requests library:
```bash
pip install requests
```
### Network Timeout
**Error**: `Request timed out. Please try again.`
**Solutions**:
* Check your internet connection
* Verify Discord is accessible (not blocked by firewall)
* Try again after a few seconds
* Implement retry logic for critical messages
```python
import time
def send_with_retry(config, max_attempts=3):
for attempt in range(max_attempts):
result = discord_tool.run(config)
if result["success"]:
return result
if attempt < max_attempts - 1:
time.sleep(2)
return result
```
### Embed Not Displaying Properly
**Problem**: Embed appears broken or missing fields
**Solutions**:
* Verify embed structure follows Discord's format
* Check field names and values are not empty
* Ensure color is a valid integer (not string)
* Verify URLs are valid and accessible
* Check character limits for each embed property
```python
# Correct embed structure
embed = {
"title": "Title",
"description": "Description",
"color": 0x5865F2, # Integer, not string
"fields": [
{"name": "Field 1", "value": "Value 1", "inline": True}
],
"footer": {"text": "Footer"}
}
```
## Security Considerations
1. **Never Commit Webhook URLs**: Add `.env` to `.gitignore`
2. **Regenerate Compromised Webhooks**: If webhook URL is exposed, regenerate it immediately
3. **Validate Input**: Sanitize user input before sending to Discord
4. **Rate Limiting**: Implement rate limiting to prevent abuse
5. **Access Control**: Restrict who can trigger Discord notifications
6. **Audit Logging**: Log notification activities for security auditing
7. **Webhook Permissions**: Use appropriate Discord channel permissions
## Dependencies
The Discord tool has minimal dependencies:
```bash
# Core dependency
pip install requests # For webhook HTTP requests
# Optional dependency
pip install jinja2 # For advanced templating
# Or install with Peargent extras
pip install peargent[discord] # Includes requests and jinja2
```
# Email Notifications
Email Notifications
Learn how to send email notifications with Peargent agents
## Overview
The Email Tool is a built-in Peargent **[Tool](/docs/tools)** that enables **[Agents](/docs/agents)** to send email notifications through SMTP or Resend API. It supports template variable substitution (Jinja2 or simple `{variable}` replacement), plain text and HTML emails, automatic format detection, and multi-provider support with intelligent fallback.
### Key Features
* **Multiple Providers** - Send via SMTP (Gmail, Outlook, custom servers) or Resend API
* **Template Support** - Jinja2 templates (when available) or simple `{variable}` replacement
* **HTML & Plain Text** - Automatic detection and handling of both formats
* **Auto-Provider Selection** - Intelligent fallback based on available credentials
* **Email Validation** - Built-in validation for sender and recipient addresses
* **Error Handling** - Comprehensive error messages and troubleshooting guidance
* **Secure Authentication** - TLS/SSL support for SMTP connections
## Common Use Cases
1. **User Notifications**: Send welcome emails, password resets, and account updates
2. **System Alerts**: Notify teams about errors, performance issues, or system events
3. **Transactional Emails**: Send receipts, invoices, and order confirmations
4. **Daily Reports**: Automate summary reports and analytics emails
5. **Event Reminders**: Send meeting reminders and calendar notifications
6. **Marketing Campaigns**: Deliver newsletters and promotional content
7. **Agent-Driven Automation**: Let AI agents send contextual emails based on user requests
8. **Multi-Language Support**: Send localized emails with template variables
## Usage with Agents
The Email **[Tool](/docs/tools)** is most powerful when integrated with **[Agents](/docs/agents)**. Agents can use the tool to automatically compose and send professional emails based on context.
### Creating an Agent with Email Tool
To use the Email tool with an agent, you need to configure it with a **[Model](/docs/models)** and pass the tool to the agent's `tools` parameter:
Before using the Email tool, you must configure your SMTP or Resend
credentials in your environment file. See
[Configuration](/docs/built-in-tools/email#configuration) for setup
instructions.
```python
from peargent import create_agent
from peargent.tools import email_tool # [!code highlight]
from peargent.models import gemini
# Create an agent with email notification capability
agent = create_agent(
name="NotificationAssistant",
description="A helpful assistant that sends email notifications",
persona=(
"You are a professional notification assistant. When asked to send emails, "
"craft clear, professional subject lines and well-formatted email bodies. "
"Use HTML formatting when beneficial for readability. Always confirm "
"successful delivery or report any errors encountered."
),
model=gemini("gemini-2.5-flash-lite"),
tools=[email_tool] # [!code highlight]
)
# Use the agent to send a notification
response = agent.run(
"Send a welcome email to alice@example.com. "
"Use welcome@company.com as the sender."
)
print(response)
```
## Examples
### Example 1: Basic Email (SMTP)
```python
from peargent.tools import email_tool
# Send a basic plain text email via SMTP
result = email_tool.run({
"to_email": "user@example.com",
"subject": "Welcome to Our Platform",
"body": "Thank you for joining! We're excited to have you on board.",
"from_email": "noreply@company.com"
})
if result["success"]:
print(f"✅ Email sent successfully!")
print(f"Provider: {result['provider']}")
if result['message_id']:
print(f"Message ID: {result['message_id']}")
else:
print(f"❌ Error: {result['error']}")
```
### Example 2: HTML Email
```python
from peargent.tools import email_tool
# Send an HTML email with rich formatting
html_body = """
Welcome to Our Platform!
Thank you for creating your account.
Next Steps:
- Complete your profile
- Explore our features
- Join our community
Get Started
"""
result = email_tool.run({
"to_email": "user@example.com",
"subject": "Welcome to Our Platform",
"body": html_body, # [!code highlight]
"from_email": "welcome@company.com"
})
if result["success"]:
print("✅ HTML email sent successfully!")
```
### Example 3: Template Variables with Jinja2
```python
from peargent.tools import email_tool
# Use Jinja2 template syntax for dynamic content
result = email_tool.run({
"to_email": "alice@example.com",
"subject": "Welcome {{ first_name }} {{ last_name }}!", # [!code highlight]
"body": """
Hello {{ first_name }},
Your account has been successfully created!
Account Details:
- Username: {{ username }}
- Email: {{ email }}
- Member Since: {{ join_date }}
Click here to activate your account:
{{ activation_link }}
Best regards,
The {{ company_name }} Team
""",
"template_vars": { # [!code highlight]
"first_name": "Alice",
"last_name": "Johnson",
"username": "alice_j",
"email": "alice@example.com",
"join_date": "January 6, 2026",
"activation_link": "https://app.example.com/activate/abc123",
"company_name": "Tech Corp"
},
"from_email": "noreply@techcorp.com"
})
if result["success"]:
print("✅ Templated email sent successfully!")
```
### Example 4: Simple Template (Without Jinja2)
```python
from peargent.tools import email_tool
# If Jinja2 is not installed, use simple {variable} syntax
result = email_tool.run({
"to_email": "bob@example.com",
"subject": "Hello {name}!", # [!code highlight]
"body": "Welcome {name}! Your order #{order_id} has been confirmed.",
"template_vars": { # [!code highlight]
"name": "Bob",
"order_id": "12345"
},
"from_email": "orders@shop.com"
})
if result["success"]:
print("✅ Simple template email sent!")
```
### Example 5: Resend Provider
```python
from peargent.tools import email_tool
# Send via Resend API instead of SMTP
result = email_tool.run({
"to_email": "customer@example.com",
"subject": "Your order has shipped!",
"body": """
📦 Your Order Has Shipped!
Great news! Your order is on its way.
Tracking Number: 1Z999AA10123456784
Track Your Package
""",
"from_email": "test@resend.dev", # Use test@resend.dev for testing
"provider": "resend" # [!code highlight]
})
if result["success"]:
print(f"✅ Email sent via Resend!")
print(f"Message ID: {result['message_id']}")
```
### Example 6: Alert Email with Template
```python
from peargent.tools import email_tool
# Send system alert with metrics
alert_data = {
"alert_type": "High CPU Usage",
"server_name": "web-server-01",
"cpu_percent": "95",
"memory_percent": "78",
"timestamp": "2026-01-06 14:30:00",
"threshold": "80",
"dashboard_url": "https://monitoring.example.com/servers/web-01"
}
result = email_tool.run({
"to_email": "devops@example.com",
"subject": "⚠️ Alert: {{ alert_type }} on {{ server_name }}", # [!code highlight]
"body": """
Alert Type: {{ alert_type }}
Server: {{ server_name }}
Time: {{ timestamp }}
Current Metrics:
- CPU Usage: {{ cpu_percent }}% (Threshold: {{ threshold }}%)
- Memory Usage: {{ memory_percent }}%
Action Required: Please investigate immediately.
View dashboard: {{ dashboard_url }}
This is an automated alert from the monitoring system.
""",
"template_vars": alert_data, # [!code highlight]
"from_email": "alerts@company.com"
})
if result["success"]:
print("🚨 Alert email sent to DevOps team!")
```
### Example 7: Invoice Email with Rich HTML
```python
from peargent.tools import email_tool
# Send styled invoice email
invoice_data = {
"customer_name": "John Doe",
"invoice_number": "INV-2026-001",
"invoice_date": "January 6, 2026",
"amount": "$149.99",
"payment_method": "Credit Card (****1234)",
"items": "Premium Plan - Annual Subscription",
"next_billing_date": "January 6, 2027",
"invoice_url": "https://billing.example.com/invoices/001"
}
result = email_tool.run({
"to_email": "john@example.com",
"subject": "Invoice {{ invoice_number }} - Payment Received", # [!code highlight]
"body": """
✅ Thank you for your payment!
Dear {{ customer_name }},
We have received your payment of {{ amount }}.
Invoice Details:
| Invoice Number: |
{{ invoice_number }} |
| Date: |
{{ invoice_date }} |
| Amount: |
{{ amount }} |
| Payment Method: |
{{ payment_method }} |
| Items: |
{{ items }} |
Next billing date: {{ next_billing_date }}
Download Invoice PDF
Thank you for your business!
""",
"template_vars": invoice_data, # [!code highlight]
"from_email": "billing@example.com"
})
if result["success"]:
print("📄 Invoice email sent!")
```
### Example 8: Agent-Driven Email Automation
```python
from peargent import create_agent
from peargent.tools import email_tool # [!code highlight]
from peargent.models import gemini
# Create an intelligent notification agent
agent = create_agent(
name="SmartNotifier",
description="An intelligent agent that sends contextual email notifications",
persona="""
You are a smart notification assistant. When asked to send notifications:
1. Craft professional, clear subject lines
2. Write concise, informative email bodies
3. Use appropriate formatting (HTML when beneficial)
4. Include relevant details and call-to-action
5. Use template variables when provided
Always use the email_tool to send emails.
""",
model=gemini("gemini-2.5-flash-lite"),
tools=[email_tool] # [!code highlight]
)
# Scenario 1: Password reset request
print("Scenario 1: Password Reset")
print("-" * 60)
response = agent.run(
"A user named Sarah Martinez (sarah@example.com) requested a password reset. "
"Send her an email with a reset link: https://app.example.com/reset/xyz789. "
"The link expires in 1 hour. Use security@example.com as sender."
)
print(f"Agent Response: {response}\n")
# Scenario 2: Daily summary report
print("Scenario 2: Daily Summary Report")
print("-" * 60)
response = agent.run(
"Send a daily summary report to admin@example.com. "
"Today's stats: 150 new users, 1,250 active sessions, $5,420 revenue. "
"Use reports@example.com as sender."
)
print(f"Agent Response: {response}\n")
# Scenario 3: Event reminder
print("Scenario 3: Event Reminder")
print("-" * 60)
response = agent.run(
"Send a reminder to team@example.com about tomorrow's product launch meeting "
"at 10 AM EST. Include Zoom link: https://zoom.us/j/123456789. "
"Use calendar@example.com as sender."
)
print(f"Agent Response: {response}")
```
### Example 9: Batch Notifications
```python
from peargent.tools import email_tool
# Send multiple notifications
recipients = [
{"email": "alice@example.com", "name": "Alice Johnson"},
{"email": "bob@example.com", "name": "Bob Smith"},
{"email": "carol@example.com", "name": "Carol Williams"}
]
print("Sending batch notifications:\n")
for recipient in recipients:
result = email_tool.run({
"to_email": recipient["email"],
"subject": "Important Update for {{ name }}", # [!code highlight]
"body": """
Hello {{ name }},
We have an important update to share with you about our new features.
Check out what's new:
- Feature A: Enhanced performance
- Feature B: New integrations
- Feature C: Improved security
Visit our blog to learn more: https://blog.example.com/updates
Best regards,
The Team
""",
"template_vars": {"name": recipient["name"]}, # [!code highlight]
"from_email": "updates@example.com"
})
if result["success"]:
print(f"✅ Email sent to {recipient['name']}")
else:
print(f"❌ Failed to send to {recipient['name']}: {result['error']}")
```
### Example 10: Error Handling
```python
from peargent.tools import email_tool
# Comprehensive error handling
def send_with_retry(email_config, max_retries=3):
"""Send email with retry logic."""
for attempt in range(max_retries):
result = email_tool.run(email_config)
if result["success"]:
print(f"✅ Email sent successfully on attempt {attempt + 1}")
return result
else:
print(f"❌ Attempt {attempt + 1} failed: {result['error']}")
# Check for specific errors
if "authentication" in result['error'].lower():
print("Authentication error - check credentials")
break
elif "invalid" in result['error'].lower() and "email" in result['error'].lower():
print("Invalid email address - check recipient")
break
elif attempt < max_retries - 1:
print(f"Retrying... ({attempt + 2}/{max_retries})")
return result
# Use the retry function
result = send_with_retry({
"to_email": "user@example.com",
"subject": "Test Email",
"body": "Testing error handling and retry logic.",
"from_email": "test@example.com"
})
```
## Parameters
The Email tool accepts the following parameters:
* **to\_email** (string, required): Recipient email address. Must be a valid email format
* **subject** (string, required): Email subject line. Supports template variables when `template_vars` is provided
* **body** (string, required): Email body content. Supports both plain text and HTML. HTML is automatically detected. Supports template variables when `template_vars` is provided
* **from\_email** (string, required): Sender email address. Must be a valid email format
* **template\_vars** (dictionary, optional): Variables for template substitution. If provided, applies Jinja2 templating (if available) or simple `{variable}` replacement to both subject and body
* **provider** (string, optional, default: "smtp"): Email provider to use. Options: "smtp" or "resend". Auto-fallback enabled based on available credentials
* **smtp\_use\_tls** (boolean, optional, default: True): Whether to use TLS encryption for SMTP connections
## Return Value
The tool returns a dictionary with the following structure:
```python
{
"success": True, # Boolean indicating success/failure
"provider": "smtp", # Provider used: "smtp" or "resend"
"message_id": "abc123...", # Message ID (Resend only, None for SMTP)
"error": None # Error message if failed, None otherwise
}
```
### Success Response Example
```python
{
"success": True,
"provider": "smtp",
"message_id": None, # SMTP doesn't provide message IDs
"error": None
}
# OR for Resend:
{
"success": True,
"provider": "resend",
"message_id": "550e8400-e29b-41d4-a716-446655440000",
"error": None
}
```
### Error Response Example
```python
{
"success": False,
"provider": "smtp",
"message_id": None,
"error": "SMTP authentication failed. Check username and password."
}
```
## Configuration
### SMTP Configuration
Set these environment variables in your `.env` file:
```bash
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USERNAME=your-email@gmail.com
SMTP_PASSWORD=your-app-password
```
#### Gmail Setup
For Gmail, you need to use an App Password:
1. Enable 2-Factor Authentication in your Google Account
2. Go to Google Account > Security > 2-Step Verification > App Passwords
3. Generate a new App Password for "Mail"
4. Use that password in `SMTP_PASSWORD`
```bash
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USERNAME=your-email@gmail.com
SMTP_PASSWORD=your-16-character-app-password
```
#### Outlook/Office 365 Setup
```bash
SMTP_HOST=smtp-mail.outlook.com
SMTP_PORT=587
SMTP_USERNAME=your-email@outlook.com
SMTP_PASSWORD=your-password
```
#### Custom SMTP Server Setup
```bash
SMTP_HOST=mail.yourdomain.com
SMTP_PORT=587 # or 465 for SSL
SMTP_USERNAME=noreply@yourdomain.com
SMTP_PASSWORD=your-password
```
### Resend Configuration
Set this environment variable in your `.env` file:
```bash
RESEND_API_KEY=re_xxxxxxxxxxxxxxxxxxxxxxxxxxxx
```
To get a Resend API key:
1. Sign up at [resend.com](https://resend.com)
2. Go to API Keys in your dashboard
3. Create a new API key
4. Add it to your `.env` file
**Note**: For production use with Resend, you need to verify your domain. For testing, you can use `test@resend.dev` as the sender address.
### Auto-Provider Selection
The tool automatically selects the best available provider:
* If only SMTP credentials are set → uses SMTP
* If only Resend API key is set → uses Resend
* If both are set → uses SMTP by default (or specify with `provider` parameter)
* If neither is set → returns error with helpful message
## Template Systems
### Jinja2 Templating (Recommended)
When Jinja2 is installed, you get full template power:
```python
# Install Jinja2
pip install jinja2
# Use advanced features
result = email_tool.run({
"to_email": "user@example.com",
"subject": "Order Confirmation",
"body": """
Hello {{ customer.name }},
Your order has been confirmed!
{% if items %}
Items ordered:
{% for item in items %}
- {{ item.name }}: ${{ item.price }}
{% endfor %}
{% endif %}
Total: ${{ total }}
{% if discount > 0 %}
You saved: ${{ discount }} with your coupon!
{% endif %}
Thank you for your purchase!
""",
"template_vars": {
"customer": {"name": "Alice"},
"items": [
{"name": "Widget A", "price": "29.99"},
{"name": "Widget B", "price": "39.99"}
],
"total": "69.98",
"discount": "10.00"
},
"from_email": "orders@shop.com"
})
```
### Simple Templating (Fallback)
Without Jinja2, use simple `{variable}` syntax:
```python
# Works without Jinja2 installation
result = email_tool.run({
"to_email": "user@example.com",
"subject": "Hello {name}!",
"body": "Welcome {name}! Your account {account_id} is ready.",
"template_vars": {
"name": "Bob",
"account_id": "12345"
},
"from_email": "welcome@app.com"
})
```
### Without Templates
To send emails without variable substitution:
```python
# Omit template_vars to send content as-is
result = email_tool.run({
"to_email": "user@example.com",
"subject": "Static subject with {{ literal }} brackets",
"body": "This {{ variable }} syntax is sent literally, not replaced.",
"from_email": "test@example.com"
# No template_vars parameter
})
```
## Best Practices
1. **Use Environment Variables**: Never hardcode credentials in your code. Always use `.env` files
2. **Validate Before Sending**: Check email addresses and required fields before calling the tool
3. **Handle Errors Gracefully**: Always check `result["success"]` and handle errors appropriately
4. **Use HTML for Rich Content**: Use HTML emails for better formatting and visual appeal
5. **Test with Different Providers**: Test your emails with both SMTP and Resend to ensure compatibility
6. **Template for Personalization**: Use template variables for dynamic, personalized content
7. **Secure Credentials**: Use App Passwords for Gmail, not your main password
8. **Monitor Delivery**: Check message IDs (Resend) and implement retry logic for critical emails
9. **Respect Rate Limits**: Be mindful of provider rate limits when sending batch emails
10. **Professional Formatting**: Use proper email etiquette and clear call-to-action buttons
## Performance Considerations
* SMTP requests typically complete in 2-5 seconds depending on server
* Resend API requests typically complete in 1-2 seconds
* Email validation is performed client-side (instant)
* HTML detection is automatic and efficient
* Template rendering adds minimal overhead (\< 100ms for most templates)
## Troubleshooting
### SMTP Authentication Error
**Error**: `SMTP authentication failed. Check username and password.`
**Solutions**:
* For Gmail: Use an App Password, not your regular password
* Verify `SMTP_USERNAME` and `SMTP_PASSWORD` are correct
* Check if 2FA is enabled (required for Gmail App Passwords)
* Ensure you're using the correct SMTP host and port
```bash
# Gmail requires App Password
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USERNAME=your-email@gmail.com
SMTP_PASSWORD=your-app-password # Not your regular password!
```
### Connection Error
**Error**: `Failed to connect to SMTP server: smtp.example.com:587`
**Solutions**:
* Verify `SMTP_HOST` and `SMTP_PORT` are correct
* Check your internet connection
* Ensure your firewall allows SMTP connections
* Try alternative ports (587 for TLS, 465 for SSL, 25 for plain)
### Invalid Email Address
**Error**: `Invalid recipient email address: invalid-email`
**Solutions**:
* Ensure email addresses follow proper format: `user@domain.com`
* Check for typos in email addresses
* Validate email addresses before passing to the tool
```python
import re
def is_valid_email(email):
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return re.match(pattern, email) is not None
# Validate before sending
if is_valid_email(recipient):
result = email_tool.run({...})
else:
print("Invalid email address!")
```
### Missing Configuration
**Error**: `Missing SMTP configuration: SMTP_HOST, SMTP_PASSWORD. Set these in .env file.`
**Solutions**:
* Create a `.env` file in your project root
* Add all required SMTP credentials
* Or configure Resend as an alternative
```bash
# Create .env file
touch .env
# Add credentials
echo "SMTP_HOST=smtp.gmail.com" >> .env
echo "SMTP_PORT=587" >> .env
echo "SMTP_USERNAME=your-email@gmail.com" >> .env
echo "SMTP_PASSWORD=your-app-password" >> .env
```
### Resend API Error
**Error**: `Resend API error: Invalid API key`
**Solutions**:
* Verify `RESEND_API_KEY` is correct
* Ensure API key starts with `re_`
* Check if API key has been revoked
* Generate a new API key if needed
```bash
# Check your API key format
RESEND_API_KEY=re_xxxxxxxxxxxxxxxxxxxxxxxxxxxx
```
### Template Rendering Issues
**Problem**: Variables not being replaced
**Solutions**:
* Ensure `template_vars` dictionary is provided
* For Jinja2: Use `{{ variable }}` syntax
* For simple: Use `{variable}` syntax
* Verify variable names match exactly (case-sensitive)
```python
# Correct usage
result = email_tool.run({
"subject": "Hello {{ name }}", # or "Hello {name}"
"body": "Welcome {{ name }}!",
"template_vars": {"name": "Alice"}, # Must provide template_vars
...
})
```
### HTML Not Rendering
**Problem**: HTML appears as plain text
**Solutions**:
* Ensure you're using proper HTML tags (``, ``, ``, etc.)
* The tool automatically detects HTML by looking for HTML tags
* Use inline styles for better email client compatibility
* Test your HTML in different email clients
```python
# HTML will be auto-detected
body = """
This is HTML
It will be sent as HTML, not plain text.
"""
```
### Network Timeout
**Error**: `Request timed out. Please try again.`
**Solutions**:
* Check your internet connection
* Try again after a few seconds
* Implement retry logic for critical emails
* Consider switching providers if timeouts persist
```python
import time
def send_with_retry(config, max_attempts=3, delay=2):
for attempt in range(max_attempts):
result = email_tool.run(config)
if result["success"]:
return result
if attempt < max_attempts - 1:
time.sleep(delay)
return result
```
## Security Considerations
1. **Never Commit Credentials**: Add `.env` to `.gitignore`
2. **Use App Passwords**: For Gmail, use App Passwords instead of your main password
3. **Rotate API Keys**: Regularly rotate your Resend API keys
4. **Validate Input**: Always validate email addresses and content
5. **TLS/SSL**: Use encrypted connections (TLS/SSL) for SMTP
6. **Rate Limiting**: Implement rate limiting for email sending
7. **Audit Logging**: Log email sending activities for security auditing
## Dependencies
The Email tool has minimal dependencies:
```bash
# Core dependencies (included in Python standard library)
# - smtplib (built-in)
# - email.mime (built-in)
# - re (built-in)
# Optional dependencies
pip install requests # For Resend API support
pip install jinja2 # For advanced templating
# Or install with Peargent extras
pip install peargent[email] # Includes requests and jinja2
```
# Built-in Tools
Built-in Tools
Ready-to-use tools that extend agent capabilities with powerful built-in
functionality
Peargent provides a growing collection of built-in **[Tools](/docs/tools)** that solve common tasks without requiring custom implementations. These tools are production-ready, well-tested, and integrate seamlessly with **[Agents](/docs/agents)**.
## Why Built-in Tools?
Built-in tools save development time and provide:
* **Zero Configuration** - Import and use immediately, no setup required
* **Production Ready** - Thoroughly tested and optimized for reliability
* **Best Practices** - Built with proper error handling, validation, and security
* **Consistent API** - Same interface patterns across all built-in tools
* **Maintained** - Regular updates and improvements from the Peargent team
## Available Built-in Tools
### Text Extraction Tool
Extract plain text and metadata from various document formats including HTML, PDF, DOCX, TXT, Markdown, and URLs. This tool enables agents to read and process content from different file types and web pages. Supported formats: HTML/XHTML, PDF, DOCX, TXT, Markdown, and URLs (with SSRF protection). **[Learn more about Text Extraction Tool →](/docs/built-in-tools/text-extraction)**
### Wikipedia Search
Search Wikipedia and extract knowledge from articles with smart fuzzy matching, summary extraction, and multi-language support. This tool enables agents to retrieve accurate information from Wikipedia articles, handle disambiguation pages, get article suggestions, and extract related links and categories. Supports all Wikipedia languages and provides intelligent handling of missing or ambiguous articles. **[Learn more about Wikipedia Search →](/docs/built-in-tools/wikipedia-search)**
### Email Notifications
Send professional email notifications through SMTP or Resend API with template variable substitution, HTML support, and intelligent provider selection. This tool enables agents to send emails with Jinja2 templating or simple variable replacement, automatic HTML detection, built-in validation, and comprehensive error handling. Supports Gmail, Outlook, custom SMTP servers, and the Resend API with automatic fallback between providers. **[Learn more about Email Notifications →](/docs/built-in-tools/email)**
### Discord Notifications
Send Discord messages and rich embeds to channels through webhooks with template variable substitution, custom branding, and comprehensive formatting support. This tool enables agents to send messages with Jinja2 templating or simple variable replacement, rich embed support with fields and images, custom usernames and avatars, and automatic webhook URL loading from environment variables. Perfect for system alerts, build notifications, deployment updates, and AI-powered Discord bots. **[Learn more about Discord Notifications →](/docs/built-in-tools/discord)**
### DateTime Operations
Work with dates, times, timezones, and perform time-based calculations with support for current time retrieval, time difference calculations, date parsing, and timezone conversions. This tool enables agents to get current UTC or timezone-specific time, calculate duration between dates with smart unit selection, parse ISO 8601 and Unix timestamps, convert between IANA timezones, and generate human-readable time descriptions. Perfect for scheduling, timezone coordination, time tracking, and natural language time queries. **[Learn more about DateTime Operations →](/docs/built-in-tools/dateTime)**
### Web Search
Search the web with Peargent agents using DuckDuckGo for real-time information retrieval. This tool provides rich search results with titles, snippets, and URLs, supporting regional filtering, safe search controls, time-based filtering, and customizable result counts. Enables agents to access up-to-date information from the web, verify claims, conduct research, and ground responses with current data. No API keys required - works out of the box with zero configuration. Perfect for research assistance, fact-checking, RAG applications, competitive intelligence, and news monitoring. **[Learn more about Web Search →](/docs/built-in-tools/websearch)**
## Coming Soon
More built-in tools are in development:
* **Image Analysis Tool** - Extract text and analyze images
* **File System Tool** - Read, write, and manage files safely
* **HTTP Request Tool** - Make API calls with built-in retry logic
Check the **[Peargent GitHub repository](https://github.com/Peargent/peargent/tree/main/peargent/tools)** for the latest updates.
# Text Extraction Tool
Text Extraction Tool
Learn how to use the text extraction tool with Peargent agents
## Overview
The Text Extraction Tool is a built-in Peargent **[Tool](/docs/tools)** that enables **[Agents](/docs/agents)** to extract plain text from various document formats. It supports HTML, PDF, DOCX, TXT, and Markdown files, as well as URLs. The tool can optionally extract metadata such as title, author, page count, and character counts.
### Supported Formats
* **HTML/XHTML** - Web pages with metadata extraction (title, description, author)
* **PDF** - PDF documents with metadata (title, author, subject, page count)
* **DOCX** - Microsoft Word documents with document properties
* **TXT** - Plain text files with automatic encoding detection
* **Markdown** - Markdown files with title extraction from headers
* **URLs** - HTTP/HTTPS web resources with built-in SSRF protection
## Common Use Cases
1. **Document Summarization**: Extract text from documents and have agents summarize them
2. **Information Extraction**: Extract specific information (emails, phone numbers, etc.) from documents
3. **Content Analysis**: Analyze document sentiment, topics, or keywords
4. **Batch Processing**: Process multiple documents programmatically
5. **Web Scraping**: Extract text from web pages while preserving structure
6. **Research Assistance**: Analyze research papers and academic documents
7. **Compliance Review**: Extract and review document contents for compliance checking
## Usage with Agents
The Text Extraction **[Tool](/docs/tools)** is most powerful when integrated with **[Agents](/docs/agents)**. Agents can use the tool to automatically extract and process document content.
### Creating an Agent with Text Extraction
To use the text extraction tool with an agent, you need to configure it with a **[Model](/docs/models)** and pass the tool to the agent's `tools` parameter:
```python
from peargent import create_agent
from peargent.tools import text_extractor # [!code highlight]
from peargent.models import gemini
# Create an agent with text extraction capability
agent = create_agent(
name="DocumentAnalyzer",
description="Analyzes documents and extracts key information",
persona=(
"You are a document analysis expert. When asked about a document, "
"use the text extraction tool to extract its content, then analyze "
"and summarize the information."
),
model=gemini("gemini-2.5-flash-lite"),
tools=[text_extractor] # [!code highlight]
)
# Use the agent to analyze a document
response = agent.run("Summarize the key points from document.pdf")
print(response)
```
## Examples
### Example 1: Extract Text with Metadata
```python
from peargent.tools import text_extractor
# Extract text and metadata from an HTML file
result = text_extractor.run({
"file_path": "article.html",
"extract_metadata": True # [!code highlight]
})
if result["success"]:
print(f"Title: {result['metadata']['title']}")
print(f"Author: {result['metadata']['author']}")
print(f"Word Count: {result['metadata']['word_count']}")
print(f"Content:\n{result['text']}")
else:
print(f"Error: {result['error']}")
```
### Example 2: Extract from URL
```python
from peargent.tools import text_extractor
# Extract text from a web page
result = text_extractor.run({
"file_path": "https://example.com/article", # [!code highlight]
"extract_metadata": True
})
if result["success"]:
print(f"Website Title: {result['metadata']['title']}")
print(f"Content: {result['text'][:500]}...")
```
### Example 3: Extract with Length Limit
```python
from peargent.tools import text_extractor
# Extract text but limit to 1000 characters
result = text_extractor.run({
"file_path": "long_document.pdf",
"extract_metadata": True,
"max_length": 1000 # [!code highlight]
})
print(f"Text (max 1000 chars): {result['text']}")
```
### Example 4: Batch Processing Multiple Files
```python
from peargent.tools import text_extractor
import os
documents = ["doc1.pdf", "doc2.docx", "doc3.html"]
for file_path in documents:
if os.path.exists(file_path):
result = text_extractor.run({
"file_path": file_path,
"extract_metadata": True
})
if result["success"]:
print(f"\n{file_path} ({result['format']})")
print(f"Words: {result['metadata'].get('word_count', 'N/A')}")
print(f"Preview: {result['text'][:150]}...")
else:
print(f"Error processing {file_path}: {result['error']}")
```
### Example 5: Agent Document Analysis
```python
from peargent import create_agent
from peargent.tools import text_extractor # [!code highlight]
from peargent.models import gemini
# Create a document analysis agent
agent = create_agent(
name="ResearchAssistant",
description="Analyzes research papers and extracts key information",
persona=(
"You are a research assistant specializing in document analysis. "
"When given a document, extract its content and identify: "
"1) Main topic, 2) Key findings, 3) Methodology, 4) Conclusions"
),
model=gemini("gemini-2.5-flash-lite"),
tools=[text_extractor] # [!code highlight]
)
# Ask the agent to analyze a research paper
response = agent.run(
"Please analyze research_paper.pdf and provide a structured summary"
)
print(response)
```
## Parameters
The text extraction tool accepts the following parameters:
* **file\_path** (string, required): Path to the file or URL to extract text from
* **extract\_metadata** (boolean, optional, default: False): Whether to extract metadata like title, author, page count, etc.
* **max\_length** (integer, optional): Maximum text length to return. If exceeded, text is truncated with "..." appended
## Return Value
The tool returns a dictionary with the following structure:
```python
{
"text": "Extracted plain text content",
"metadata": {
"title": "Document Title",
"author": "Author Name",
# ... additional metadata depending on format
},
"format": "pdf", # Detected file format
"success": True,
"error": None
}
```
## Metadata by Format
Different document formats provide different metadata:
**HTML/XHTML:**
* `title` - Page title
* `description` - Meta description tag
* `author` - Meta author tag
* `word_count` - Number of words
* `char_count` - Number of characters
**PDF:**
* `title` - Document title
* `author` - Document author
* `subject` - Document subject
* `creator` - Application that created the PDF
* `producer` - PDF producer
* `creation_date` - When the document was created
* `page_count` - Total number of pages
* `word_count` - Total word count
* `char_count` - Total character count
**DOCX:**
* `title` - Document title
* `author` - Document author
* `subject` - Document subject
* `created` - Creation date and time
* `modified` - Last modification date and time
* `word_count` - Total word count
* `char_count` - Total character count
* `paragraph_count` - Number of paragraphs
**TXT/Markdown:**
* `encoding` - Text encoding used
* `word_count` - Total word count
* `char_count` - Total character count
* `line_count` - Total line count
* `title` - (Markdown only) Title extracted from first heading
## Troubleshooting
### ImportError for document libraries
If you encounter ImportError when extracting specific formats, install the required dependencies:
```bash
# For all formats
pip install peargent[text-extraction]
# Or individually
pip install beautifulsoup4 pypdf python-docx
```
### SSRF Protection Errors
If you receive "Access to localhost is not allowed" error, ensure you're using a public URL:
```python
# This will fail
result = text_extractor.run({"file_path": "http://localhost:8000/doc"})
# Use a public URL instead
result = text_extractor.run({"file_path": "https://example.com/doc"})
```
### Encoding Issues with Text Files
For text files with non-standard encoding, the tool automatically detects encoding. If issues persist, ensure the file is properly encoded.
# Web Search
Web Search
Search the web with Peargent agents using DuckDuckGo
## Overview
The Web Search Tool is a built-in Peargent **[Tool](/docs/tools)** that enables **[Agents](/docs/agents)** to search the web for up-to-date information using DuckDuckGo. It provides search results with titles, snippets, and URLs, supporting regional filtering, safe search controls, time-based filtering, and customizable result counts. This tool is essential for grounding agent responses with current data and enabling real-time information retrieval.
### Key Features
* **Real-Time Search** - Access up-to-date information from the web via DuckDuckGo
* **Rich Results** - Get titles, snippets, and URLs for each search result
* **Regional Filtering** - Localize search results by region (US, UK, Germany, etc.)
* **Safe Search** - Control content filtering with strict, moderate, or off settings
* **Time-Based Filtering** - Filter results by day, week, month, or year
* **Customizable Results** - Control the number of results returned (1-25)
* **RAG Integration** - Perfect for Retrieval Augmented Generation workflows
* **Zero Configuration** - No API keys required, works out of the box
## Common Use Cases
1. **Research & Fact-Checking**: Verify claims and gather authoritative information
2. **Real-Time Information**: Get current news, events, and developments
3. **RAG Applications**: Retrieve relevant documents for context-aware responses
4. **Market Research**: Gather competitive intelligence and industry insights
5. **Content Creation**: Research topics for articles, blogs, and reports
6. **Data Gathering**: Collect information for analysis and decision-making
7. **Educational Queries**: Find tutorials, guides, and learning resources
8. **Trend Analysis**: Monitor emerging trends and topics
## Usage with Agents
The Web Search **[Tool](/docs/tools)** is most powerful when integrated with **[Agents](/docs/agents)**. Agents can use the tool to automatically search the web and synthesize information based on context.
### Creating an Agent with Web Search Tool
To use the Web Search tool with an agent, you need to configure it with a **[Model](/docs/models)** and pass the tool to the agent's `tools` parameter:
The Web Search tool uses DuckDuckGo and requires the `ddgs` library. Install
it with: `pip install peargent[web-search]`
```python
from peargent import create_agent
from peargent.tools import websearch_tool # [!code highlight]
from peargent.models import gemini
# Create an agent with web search capability
agent = create_agent(
name="ResearchAssistant",
description="A helpful research assistant with web search capabilities",
persona=(
"You are a professional research assistant. When asked questions, "
"use the web search tool to find current, authoritative information. "
"Provide comprehensive, well-researched answers with proper citations. "
"Always cite your sources with URLs and verify facts from multiple sources."
),
model=gemini("gemini-2.5-flash-lite"),
tools=[websearch_tool] # [!code highlight]
)
# Use the agent to research a topic
response = agent.run(
"What are the latest developments in renewable energy technology in 2026?"
)
print(response)
```
## Examples
### Example 1: Basic Web Search
```python
from peargent.tools import websearch_tool
# Perform a simple web search
result = websearch_tool.run({
"query": "Python programming tutorials"
})
if result["success"] and result["results"]:
print(f"✅ Found {result['metadata']['result_count']} results")
print(f"Search engine: {result['metadata']['search_engine']}")
# Display top results
for i, r in enumerate(result["results"][:3], 1):
print(f"\n{i}. {r['title']}")
print(f" URL: {r['url']}")
print(f" {r['snippet'][:150]}...")
else:
print(f"❌ Error: {result['error']}")
```
### Example 2: Limit Number of Results
```python
from peargent.tools import websearch_tool
# Search with limited results
result = websearch_tool.run({
"query": "artificial intelligence trends", # [!code highlight]
"max_results": 3 # [!code highlight]
})
if result["success"] and result["results"]:
print(f"✅ Found {len(result['results'])} results\n")
for i, r in enumerate(result["results"], 1):
print(f"{i}. {r['title']}")
print(f" {r['url']}\n")
```
### Example 3: Regional Search Filtering
```python
from peargent.tools import websearch_tool
# Search with regional filtering
result = websearch_tool.run({
"query": "best restaurants near me",
"max_results": 5,
"region": "us-en" # [!code highlight]
})
if result["success"] and result["results"]:
print(f"Region: {result['metadata']['region']}")
print(f"✅ Found {len(result['results'])} results\n")
for i, r in enumerate(result["results"], 1):
print(f"{i}. {r['title']}")
print(f" {r['url']}")
```
### Example 4: Time-Based Filtering
```python
from peargent.tools import websearch_tool
# Search for recent content (past week)
result = websearch_tool.run({
"query": "technology news",
"max_results": 5,
"time_range": "w" # [!code highlight]
})
if result["success"] and result["results"]:
print(f"✅ Found {len(result['results'])} recent results")
if 'time_range' in result['metadata']:
print(f"Time range: {result['metadata']['time_range']}")
for i, r in enumerate(result["results"], 1):
print(f"\n{i}. {r['title']}")
print(f" {r['snippet'][:100]}...")
print(f" Source: {r['url']}")
```
### Example 5: Safe Search Settings
```python
from peargent.tools import websearch_tool
# Search with strict safe search
result = websearch_tool.run({
"query": "educational content for children",
"max_results": 5,
"safesearch": "strict" # [!code highlight]
})
if result["success"] and result["results"]:
print(f"Safe search: {result['metadata']['safesearch']}")
print(f"✅ Found {len(result['results'])} safe results\n")
for r in result["results"]:
print(f"- {r['title']}")
print(f" {r['url']}\n")
```
### Example 6: Multi-Query Research Pattern
```python
from peargent.tools import websearch_tool
# RAG-style information retrieval with multiple queries
queries = [ # [!code highlight]
"artificial intelligence ethics",
"AI bias and fairness",
"responsible AI development"
]
all_sources = []
for query in queries: # [!code highlight]
result = websearch_tool.run({
"query": query,
"max_results": 3
})
if result["success"] and result["results"]:
all_sources.extend(result["results"])
print(f"✅ {query}: {len(result['results'])} results")
print(f"\n✅ Gathered {len(all_sources)} total sources across {len(queries)} queries\n")
# Display unique sources
print("Sample sources:")
for i, source in enumerate(all_sources[:5], 1):
print(f"{i}. {source['title']}")
print(f" {source['url']}\n")
```
### Example 7: Recent News Search
```python
from peargent.tools import websearch_tool
# Search for breaking news (past day)
result = websearch_tool.run({
"query": "breaking news technology",
"max_results": 5,
"time_range": "d" # [!code highlight]
})
if result["success"] and result["results"]:
print("🔥 Recent articles:\n")
for i, r in enumerate(result["results"], 1):
print(f"{i}. {r['title']}")
print(f" {r['snippet'][:100]}...")
print(f" Source: {r['url']}\n")
```
### Example 8: International Search
```python
from peargent.tools import websearch_tool
# Search in different languages/regions
regions = { # [!code highlight]
"us-en": "best tech startups",
"uk-en": "best tech startups",
"de-de": "beste Tech-Startups"
}
for region, query in regions.items(): # [!code highlight]
result = websearch_tool.run({
"query": query,
"max_results": 2,
"region": region # [!code highlight]
})
if result["success"] and result["results"]:
print(f"\n{region.upper()} Results:")
for r in result["results"]:
print(f"- {r['title']}")
```
### Example 9: Comprehensive Research Agent
```python
from peargent import create_agent
from peargent.tools import websearch_tool # [!code highlight]
from peargent.models import gemini
# Create an expert research agent
research_agent = create_agent(
name="DeepResearchAgent",
description="An expert research assistant with web search capabilities",
persona="""You are an expert research assistant with web search capabilities.
When researching a topic:
1. Break down complex questions into specific search queries
2. Search for current, authoritative information
3. Cross-reference multiple sources
4. Synthesize findings into clear, comprehensive answers
5. Always cite your sources with URLs
Be thorough, accurate, and provide evidence-based responses.""",
model=gemini("gemini-2.5-flash-lite"),
tools=[websearch_tool] # [!code highlight]
)
# Complex research query
query = "Compare the latest advancements in solar energy vs wind energy in 2026"
print(f"Research query: {query}\n")
print("Agent researching...\n")
response = research_agent.run(query)
print(f"Research findings:\n{response}")
```
### Example 10: Fact-Checking Agent
```python
from peargent import create_agent
from peargent.tools import websearch_tool # [!code highlight]
from peargent.models import gemini
# Create a fact-checking agent
fact_checker = create_agent(
name="FactChecker",
description="A fact-checking assistant with web search",
persona="""You are a fact-checking assistant. When given a claim:
1. Search for authoritative sources
2. Look for recent information
3. Verify facts from multiple angles
4. Provide a verdict: True, False, Partially True, or Unverified
5. Always cite sources with URLs""",
model=gemini("gemini-2.5-flash-lite"),
tools=[websearch_tool] # [!code highlight]
)
# Claim to verify
claim = "Python is the most popular programming language in 2026"
print(f"Claim to verify: {claim}\n")
verification = fact_checker.run(f"Fact-check this claim: {claim}")
print(f"Verification result:\n{verification}")
```
### Example 11: News Monitoring Agent
```python
from peargent import create_agent
from peargent.tools import websearch_tool # [!code highlight]
from peargent.models import gemini
# Create a news monitoring agent
news_agent = create_agent(
name="NewsMonitor",
description="A news monitoring assistant",
persona="""You are a news monitoring assistant. When asked about current events:
1. Search for the most recent news
2. Focus on credible news sources
3. Provide a balanced summary
4. Include publication dates when available
5. Cite all sources""",
model=gemini("gemini-2.5-flash-lite"),
tools=[websearch_tool] # [!code highlight]
)
# Monitor specific topic
topic = "quantum computing breakthroughs"
print(f"Monitoring news for: {topic}\n")
news_summary = news_agent.run(
f"What are the latest news and developments about {topic}? "
"Focus on articles from the past week."
)
print(f"News Summary:\n{news_summary}")
```
### Example 12: Academic Research Agent
```python
from peargent import create_agent
from peargent.tools import websearch_tool # [!code highlight]
from peargent.models import gemini
# Create an academic research agent
academic_agent = create_agent(
name="AcademicResearcher",
description="An academic research assistant",
persona="""You are an academic research assistant. When researching:
1. Search for scholarly articles and research papers
2. Focus on peer-reviewed sources when possible
3. Look for recent publications and studies
4. Provide comprehensive literature reviews
5. Include proper citations with URLs
6. Identify research gaps and controversies""",
model=gemini("gemini-2.5-flash-lite"),
tools=[websearch_tool] # [!code highlight]
)
# Research topic
topic = "machine learning applications in healthcare"
print(f"Academic research on: {topic}\n")
literature_review = academic_agent.run(
f"Provide a comprehensive overview of recent research on {topic}. "
"Include key findings, methodologies, and future directions."
)
print(f"Literature Review:\n{literature_review}")
```
### Example 13: Competitive Intelligence Agent
```python
from peargent import create_agent
from peargent.tools import websearch_tool # [!code highlight]
from peargent.models import gemini
# Create a competitive intelligence agent
competitor_agent = create_agent(
name="CompetitorAnalyst",
description="A competitive intelligence analyst",
persona="""You are a competitive intelligence analyst. When analyzing competitors:
1. Search for recent company news and announcements
2. Identify product launches and features
3. Analyze market positioning and strategies
4. Look for financial performance indicators
5. Track industry trends and disruptions
6. Provide actionable insights with sources""",
model=gemini("gemini-2.5-flash-lite"),
tools=[websearch_tool] # [!code highlight]
)
# Analyze competitor
company = "OpenAI"
print(f"Analyzing competitor: {company}\n")
analysis = competitor_agent.run(
f"Provide a competitive intelligence report on {company}. "
"Include recent developments, product launches, and market strategy."
)
print(f"Competitive Analysis:\n{analysis}")
```
### Example 14: Search Result Processing
```python
from peargent.tools import websearch_tool
# Search and process results
result = websearch_tool.run({
"query": "Python web frameworks 2026",
"max_results": 10
})
if result["success"] and result["results"]:
# Extract domains
domains = [r['url'].split('/')[2] for r in result["results"]]
unique_domains = set(domains)
print(f"✅ Found results from {len(unique_domains)} unique domains\n")
# Group by domain
by_domain = {}
for r in result["results"]:
domain = r['url'].split('/')[2]
if domain not in by_domain:
by_domain[domain] = []
by_domain[domain].append(r)
# Display grouped results
for domain, results in by_domain.items():
print(f"\n{domain} ({len(results)} results):")
for r in results:
print(f" - {r['title']}")
```
### Example 15: Error Handling and Retry Logic
```python
from peargent.tools import websearch_tool
import time
def search_with_retry(query, max_retries=3, delay=2):
"""Perform web search with retry logic."""
for attempt in range(max_retries):
result = websearch_tool.run({"query": query})
if result["success"]:
print(f"✅ Search successful on attempt {attempt + 1}")
return result
else:
print(f"❌ Attempt {attempt + 1} failed: {result['error']}")
# Check for specific errors
if "timed out" in result['error'].lower():
print("Request timed out - waiting before retry")
time.sleep(delay * 2)
elif "ddgs library" in result['error'].lower():
print("Missing dependency - cannot retry")
break
elif attempt < max_retries - 1:
print(f"Retrying... ({attempt + 2}/{max_retries})")
time.sleep(delay)
return result
# Use the retry function
result = search_with_retry("artificial intelligence news")
if result["success"] and result["results"]:
print(f"\n✅ Found {len(result['results'])} results")
```
### Example 16: Batch Search with Different Parameters
```python
from peargent.tools import websearch_tool
# Batch search with different configurations
search_configs = [ # [!code highlight]
{
"name": "Recent AI News",
"query": "artificial intelligence",
"max_results": 5,
"time_range": "w"
},
{
"name": "Academic Research",
"query": "machine learning research papers",
"max_results": 5,
"safesearch": "strict"
},
{
"name": "Tech Startups (US)",
"query": "tech startups",
"max_results": 5,
"region": "us-en"
},
{
"name": "Tech Startups (UK)",
"query": "tech startups",
"max_results": 5,
"region": "uk-en"
}
]
print("Performing batch searches:\n")
for config in search_configs: # [!code highlight]
name = config.pop("name")
result = websearch_tool.run(config)
if result["success"] and result["results"]:
print(f"✅ {name}: {len(result['results'])} results")
else:
print(f"❌ {name}: {result['error']}")
print("\n✅ Batch search completed")
```
### Example 17: RAG Pipeline Example
```python
from peargent.tools import websearch_tool
def rag_search(query, max_sources=10):
"""
Perform RAG-style search and prepare context for LLM.
Args:
query: Search query
max_sources: Maximum number of sources to retrieve
Returns:
Dictionary with context and source metadata
"""
# Search for relevant information
result = websearch_tool.run({
"query": query,
"max_results": max_sources
})
if not result["success"] or not result["results"]:
return {
"context": "",
"sources": [],
"error": result.get("error")
}
# Build context from search results
context_parts = []
sources = []
for i, r in enumerate(result["results"], 1):
# Format as context
context_parts.append(
f"Source {i}: {r['title']}\n"
f"URL: {r['url']}\n"
f"Content: {r['snippet']}\n"
)
# Store source metadata
sources.append({
"id": i,
"title": r['title'],
"url": r['url']
})
return {
"context": "\n".join(context_parts),
"sources": sources,
"query": query,
"source_count": len(sources)
}
# Use RAG search
rag_data = rag_search("benefits of renewable energy", max_sources=5)
if rag_data["context"]:
print(f"✅ Retrieved {rag_data['source_count']} sources\n")
print("Context for LLM:")
print("-" * 60)
print(rag_data["context"])
print("-" * 60)
print("\nSources:")
for source in rag_data["sources"]:
print(f"{source['id']}. {source['title']}")
print(f" {source['url']}\n")
```
## Parameters
The Web Search tool accepts the following parameters:
* **query** (string, required): Search query string. Cannot be empty or whitespace-only
* **max\_results** (integer, optional): Maximum number of results to return. Default: `5`. Range: `1-25`. Values outside this range are automatically clamped
* **region** (string, optional): Region code for localized results. Default: `"wt-wt"` (worldwide). Examples:
* `"us-en"` - United States (English)
* `"uk-en"` - United Kingdom (English)
* `"de-de"` - Germany (German)
* `"fr-fr"` - France (French)
* `"es-es"` - Spain (Spanish)
* `"ja-jp"` - Japan (Japanese)
* `"zh-cn"` - China (Chinese)
* **safesearch** (string, optional): Safe search filtering level. Default: `"moderate"`. Options:
* `"strict"` - Maximum filtering, suitable for children
* `"moderate"` - Balanced filtering (default)
* `"off"` - No filtering
* **time\_range** (string, optional): Filter results by time period. Default: `None` (all time). Options:
* `"d"` - Past day
* `"w"` - Past week
* `"m"` - Past month
* `"y"` - Past year
* `None` - All time (no filtering)
## Return Value
The tool returns a dictionary with the following structure:
```python
{
"success": True, # Boolean indicating success/failure
"results": [ # List of search results
{
"title": "Result Title",
"snippet": "Brief description or excerpt from the page",
"url": "https://example.com/page"
}
],
"metadata": { # Search metadata
"query": "search query",
"result_count": 5,
"search_engine": "DuckDuckGo",
"region": "wt-wt",
"safesearch": "moderate",
"time_range": "w" # Only present if time_range was specified
},
"error": None # Error message if failed, None otherwise
}
```
### Success Response Example
```python
{
"success": True,
"results": [
{
"title": "Python Programming Tutorial",
"snippet": "Learn Python programming with this comprehensive guide...",
"url": "https://example.com/python-tutorial"
},
{
"title": "Advanced Python Techniques",
"snippet": "Explore advanced Python programming concepts...",
"url": "https://example.com/advanced-python"
}
],
"metadata": {
"query": "Python programming tutorials",
"result_count": 2,
"search_engine": "DuckDuckGo",
"region": "wt-wt",
"safesearch": "moderate"
},
"error": None
}
```
### No Results Response Example
```python
{
"success": True,
"results": [],
"metadata": {
"query": "extremely specific query with no results",
"result_count": 0,
"search_engine": "DuckDuckGo",
"message": "No results found for your query"
},
"error": None
}
```
### Error Response Example
```python
{
"success": False,
"results": [],
"metadata": {},
"error": "Request timed out. Please try again."
}
```
## Configuration
### Installation
The Web Search tool requires the `ddgs` (DuckDuckGo Search) library:
```bash
# Install the required dependency
pip install ddgs
# Or install with Peargent extras
pip install peargent[web-search]
```
### No API Keys Required
Unlike many search APIs, the Web Search tool requires no API keys or configuration. It works out of the box once the `ddgs` library is installed.
```python
from peargent.tools import websearch_tool
# Ready to use immediately
result = websearch_tool.run({"query": "Python tutorials"})
```
## Region Codes Reference
Common region codes for localized search:
### North America
* `us-en` - United States (English)
* `ca-en` - Canada (English)
* `ca-fr` - Canada (French)
* `mx-es` - Mexico (Spanish)
### Europe
* `uk-en` - United Kingdom
* `de-de` - Germany
* `fr-fr` - France
* `es-es` - Spain
* `it-it` - Italy
* `nl-nl` - Netherlands
* `pl-pl` - Poland
* `ru-ru` - Russia
* `se-sv` - Sweden
### Asia Pacific
* `jp-jp` - Japan
* `kr-kr` - South Korea
* `cn-zh` - China
* `tw-zh` - Taiwan
* `hk-zh` - Hong Kong
* `in-en` - India
* `au-en` - Australia
* `nz-en` - New Zealand
### South America
* `br-pt` - Brazil
* `ar-es` - Argentina
* `cl-es` - Chile
### Middle East & Africa
* `il-he` - Israel
* `tr-tr` - Turkey
* `za-en` - South Africa
### Worldwide
* `wt-wt` - Worldwide (default, no regional filtering)
## Best Practices
1. **Use Descriptive Queries**: Craft specific, clear search queries for better results
2. **Limit Results Appropriately**: Request only the number of results you need (1-25)
3. **Implement Error Handling**: Always check `result["success"]` before processing results
4. **Use Time Filters for News**: Apply `time_range` when searching for recent information
5. **Regional Filtering**: Use `region` parameter for location-specific queries
6. **Safe Search for Sensitive Content**: Enable `safesearch: "strict"` when appropriate
7. **Retry Logic**: Implement retry mechanisms for network failures
8. **Source Verification**: Cross-reference multiple sources for important information
9. **Rate Limiting**: Add delays between searches to avoid overwhelming DuckDuckGo
10. **Parse Snippets Carefully**: Snippets may be truncated; visit URLs for full content
11. **Cache Results**: Store search results to avoid redundant queries
12. **Agent Personas**: Design agent personas that encourage proper source citation
## Performance Considerations
* Search requests typically complete in 1-3 seconds
* Network latency affects response time
* `max_results` parameter impacts processing time linearly
* DuckDuckGo rate limits may apply for excessive requests (implement delays)
* Regional searches may be slightly slower than worldwide searches
* Time-filtered searches perform similarly to unfiltered searches
* Consider caching results for frequently repeated queries
## Troubleshooting
### Missing ddgs Library
**Error**: `ddgs library is required for web search. Install it with: pip install ddgs`
**Solutions**:
* Install the ddgs library:
```bash
pip install ddgs
```
* Or install with Peargent extras:
```bash
pip install peargent[web-search]
```
### Empty Query Error
**Error**: `Query cannot be empty`
**Solutions**:
* Ensure query parameter is provided and not empty
* Verify query is not just whitespace
```python
# Incorrect
result = websearch_tool.run({"query": ""})
result = websearch_tool.run({"query": " "})
# Correct
result = websearch_tool.run({"query": "Python tutorials"})
```
### Network Timeout
**Error**: `Request timed out. Please try again.`
**Solutions**:
* Check your internet connection
* Verify DuckDuckGo is accessible (not blocked by firewall/proxy)
* Implement retry logic with delays
* Try again after a few seconds
```python
import time
def search_with_retry(query, max_attempts=3):
for attempt in range(max_attempts):
result = websearch_tool.run({"query": query})
if result["success"]:
return result
if attempt < max_attempts - 1:
time.sleep(2) # Wait before retry
return result
```
### No Results Found
**Problem**: Search returns no results (`result["results"]` is empty)
**Solutions**:
* Try broader, less specific search terms
* Remove quotes or special characters
* Check spelling of query terms
* Try different regional settings
* Remove or adjust time\_range filter
```python
# Too specific (may return no results)
result = websearch_tool.run({
"query": "extremely specific technical term with typos"
})
# Better (more likely to return results)
result = websearch_tool.run({
"query": "technical term general concept"
})
```
### Network Connection Error
**Error**: `Network error: [connection details]`
**Solutions**:
* Verify internet connectivity
* Check firewall/proxy settings
* Ensure DuckDuckGo is not blocked
* Try different network connection
* Check if VPN is interfering
### Invalid Parameter Values
**Problem**: Parameters not working as expected
**Solutions**:
* Verify parameter types (query: string, max\_results: int, etc.)
* Ensure `safesearch` is one of: "strict", "moderate", "off"
* Ensure `time_range` is one of: "d", "w", "m", "y", or None
* Check region code format (e.g., "us-en", not "us" or "en-us")
```python
# Incorrect
result = websearch_tool.run({
"query": "AI news",
"max_results": "5", # Should be int, not string
"safesearch": "high", # Invalid value
"time_range": "week" # Should be "w", not "week"
})
# Correct
result = websearch_tool.run({
"query": "AI news",
"max_results": 5,
"safesearch": "strict",
"time_range": "w"
})
```
### Rate Limiting
**Problem**: Searches fail after multiple rapid requests
**Solutions**:
* Implement delays between searches (1-2 seconds recommended)
* Use batch processing with delays
* Cache results to avoid redundant queries
* Limit concurrent search requests
```python
import time
# Batch search with delays
queries = ["query1", "query2", "query3"]
for query in queries:
result = websearch_tool.run({"query": query})
# Process result...
time.sleep(1) # Wait 1 second between searches
```
### DuckDuckGo Service Issues
**Problem**: Searches consistently failing
**Solutions**:
* Check DuckDuckGo status ([https://duckduckgo.com](https://duckduckgo.com))
* Try again later if service is down
* Verify ddgs library is up to date:
```bash
pip install --upgrade ddgs
```
### Results Quality Issues
**Problem**: Search results are not relevant
**Solutions**:
* Refine query with more specific terms
* Use quotes for exact phrase matching: `"exact phrase"`
* Add contextual keywords to narrow results
* Try different regional settings
* Use time\_range to filter outdated content
```python
# Vague query
result = websearch_tool.run({"query": "python"})
# More specific query
result = websearch_tool.run({
"query": "python web framework django tutorial",
"time_range": "m" # Recent content
})
```
## Security Considerations
1. **Query Sanitization**: Validate and sanitize user-provided queries before searching
2. **Result Validation**: Verify URLs in results before visiting or displaying to users
3. **Content Filtering**: Use appropriate `safesearch` settings for your use case
4. **Data Privacy**: Be aware that search queries are sent to DuckDuckGo
5. **Malicious URLs**: Results may contain malicious links; validate before accessing
6. **User Input**: Sanitize user input to prevent injection attacks
7. **Rate Limiting**: Implement rate limiting to prevent abuse
8. **Error Messages**: Avoid exposing sensitive information in error messages
9. **Logging**: Log search activities for security auditing and monitoring
10. **Access Control**: Restrict who can trigger web searches in production systems
## Dependencies
The Web Search tool has minimal dependencies:
```bash
# Core dependency (required)
pip install ddgs # For DuckDuckGo search functionality
# Or install with Peargent extras
pip install peargent[web-search]
```
**Note**: Unlike many search APIs, no API keys or authentication is required. The tool uses DuckDuckGo's public search interface.
# Wikipedia Search
Wikipedia Search
Learn how to use the Wikipedia search tool with Peargent agents
## Overview
The Wikipedia Search Tool is a built-in Peargent **[Tool](/docs/tools)** that enables **[Agents](/docs/agents)** to search Wikipedia and extract knowledge from articles. It supports article search with fuzzy matching, summary extraction, related links, categories, multi-language support, and intelligent handling of disambiguation pages and missing articles.
### Key Features
* **Smart Article Search** - Finds best match even with fuzzy queries
* **Summary Extraction** - Extracts article introductions and summaries
* **Related Links** - Optionally extracts internal Wikipedia links
* **Categories** - Optionally extracts article categories
* **Disambiguation Handling** - Provides options when articles have multiple meanings
* **Article Suggestions** - Suggests alternatives when articles don't exist
* **Multi-Language Support** - Query Wikipedia in any language (en, fr, es, de, etc.)
* **Summary Length Control** - Limit summary length to fit your needs
## Common Use Cases
1. **Research Assistance**: Quickly look up facts, definitions, and summaries for research
2. **Educational Content**: Retrieve accurate information for learning materials
3. **Content Generation**: Gather background information for articles or blog posts
4. **Fact Checking**: Verify information against Wikipedia articles
5. **Multi-Language Research**: Access Wikipedia content in various languages
6. **Topic Exploration**: Discover related topics through article links and categories
7. **Chatbot Knowledge**: Power conversational AI with Wikipedia knowledge
8. **Automated Summarization**: Extract concise summaries from Wikipedia articles
## Usage with Agents
The Wikipedia **[Tool](/docs/tools)** is most powerful when integrated with **[Agents](/docs/agents)**. Agents can use the tool to automatically search Wikipedia and retrieve accurate information.
### Creating an Agent with Wikipedia Tool
To use the Wikipedia tool with an agent, you need to configure it with a **[Model](/docs/models)** and pass the tool to the agent's `tools` parameter:
```python
from peargent import create_agent
from peargent.tools import wikipedia_tool # [!code highlight]
from peargent.models import gemini
# Create an agent with Wikipedia search capability
agent = create_agent(
name="ResearchAssistant",
description="A helpful research assistant that uses Wikipedia",
persona=(
"You are a knowledgeable research assistant. When asked about topics, "
"use Wikipedia to find accurate information and provide concise, "
"well-cited summaries. Always include the Wikipedia article title and URL."
),
model=gemini("gemini-2.5-flash-lite"),
tools=[wikipedia_tool] # [!code highlight]
)
# Use the agent to research a topic
response = agent.run("Tell me about quantum computing")
print(response)
```
## Examples
### Example 1: Basic Wikipedia Search
```python
from peargent.tools import wikipedia_tool
# Search for a well-known article
result = wikipedia_tool.run({"query": "Artificial Intelligence"})
if result["success"] and result["text"]:
print(f"Title: {result['metadata']['title']}")
print(f"URL: {result['metadata']['url']}")
print(f"\nSummary:\n{result['text']}")
# Access related links
if 'links' in result['metadata']:
print(f"\nRelated Topics: {result['metadata']['links'][:5]}")
else:
print(f"Error: {result['error']}")
```
### Example 2: Search with Categories
```python
from peargent.tools import wikipedia_tool
# Extract article categories
result = wikipedia_tool.run({
"query": "Machine Learning",
"extract_categories": True # [!code highlight]
})
if result["success"] and result["text"]:
print(f"Title: {result['metadata']['title']}")
print(f"Summary: {result['text'][:200]}...")
# Display categories
if 'categories' in result['metadata']:
print(f"\nCategories:")
for category in result['metadata']['categories'][:10]:
print(f" - {category}")
```
### Example 3: Handling Disambiguation Pages
```python
from peargent.tools import wikipedia_tool
# Search for an ambiguous term
result = wikipedia_tool.run({"query": "Mercury"})
if 'disambiguation' in result.get('metadata', {}):
print(f"'{result['metadata']['title']}' has multiple meanings:")
print("\nOptions:")
for i, option in enumerate(result['metadata']['disambiguation'][:10], 1):
print(f" {i}. {option}")
# Search for a specific option
specific_result = wikipedia_tool.run({"query": "Mercury (planet)"}) # [!code highlight]
print(f"\nSpecific article: {specific_result['metadata']['title']}")
print(f"Summary: {specific_result['text'][:150]}...")
```
### Example 4: Multi-Language Support
```python
from peargent.tools import wikipedia_tool
# Search Wikipedia in French
result = wikipedia_tool.run({
"query": "Tour Eiffel",
"language": "fr" # [!code highlight]
})
if result["success"] and result["text"]:
print(f"Title: {result['metadata']['title']}")
print(f"URL: {result['metadata']['url']}")
print(f"Summary (French): {result['text'][:200]}...")
# Search Wikipedia in Spanish
result = wikipedia_tool.run({
"query": "Inteligencia Artificial",
"language": "es" # [!code highlight]
})
if result["success"] and result["text"]:
print(f"Title: {result['metadata']['title']}")
print(f"Summary (Spanish): {result['text'][:200]}...")
```
### Example 5: Summary Length Control
```python
from peargent.tools import wikipedia_tool
# Limit summary to 300 characters
result = wikipedia_tool.run({
"query": "Climate Change",
"max_summary_length": 300 # [!code highlight]
})
if result["success"] and result["text"]:
print(f"Title: {result['metadata']['title']}")
print(f"Brief Summary: {result['text']}")
# Output text is limited to at most 300 characters, with "..." appended if truncated
```
### Example 6: Handling Non-Existent Articles
```python
from peargent.tools import wikipedia_tool
# Search for an article that doesn't exist
result = wikipedia_tool.run({"query": "asdfghjkl"})
if not result["text"]:
print("Article not found!")
# Check for suggestions
if 'suggestions' in result.get('metadata', {}):
print("\nSuggested articles:")
for suggestion in result['metadata']['suggestions']:
print(f" - {suggestion}")
if 'message' in result.get('metadata', {}):
print(f"\nMessage: {result['metadata']['message']}")
```
### Example 7: Research Agent with Wikipedia
```python
from peargent import create_agent
from peargent.tools import wikipedia_tool # [!code highlight]
from peargent.models import gemini
# Create a research agent
agent = create_agent(
name="WikipediaExpert",
description="An expert Wikipedia researcher",
persona=(
"You are an expert research assistant. Use Wikipedia to find accurate "
"information about topics. When you find information, cite the Wikipedia "
"article by including the article title and URL in your response. "
"If a term has multiple meanings (disambiguation), explain the options "
"and ask the user which one they're interested in."
),
model=gemini("gemini-2.5-flash-lite"),
tools=[wikipedia_tool] # [!code highlight]
)
# Research a complex topic
response = agent.run(
"I want to learn about quantum computing. Please provide an overview "
"and tell me about key pioneers in this field."
)
print(response)
# Follow-up question
response = agent.run(
"What are the main applications of quantum computing?"
)
print(response)
```
### Example 8: Batch Research Multiple Topics
```python
from peargent.tools import wikipedia_tool
# Research multiple topics
topics = [
"Neural Networks",
"Deep Learning",
"Natural Language Processing"
]
print("Researching multiple topics from Wikipedia:\n")
for topic in topics:
result = wikipedia_tool.run({
"query": topic,
"max_summary_length": 150
})
if result["success"] and result["text"]:
print(f"📖 {result['metadata']['title']}")
print(f" {result['text']}")
print(f" 🔗 {result['metadata']['url']}\n")
else:
print(f"❌ Could not find: {topic}\n")
```
## Parameters
The Wikipedia tool accepts the following parameters:
* **query** (string, required): Search term or article title to look up on Wikipedia
* **extract\_links** (boolean, optional, default: True): Whether to extract internal Wikipedia links from the article. Automatically limited to 50 links maximum
* **extract\_categories** (boolean, optional, default: False): Whether to extract article categories. Automatically limited to 50 categories maximum
* **max\_summary\_length** (integer, optional): Maximum summary length in characters. If exceeded, text is truncated with "..." appended
* **language** (string, optional, default: "en"): Wikipedia language code (e.g., "en" for English, "fr" for French, "es" for Spanish, "de" for German)
## Return Value
The tool returns a dictionary with the following structure:
```python
{
"text": "Article summary/introduction text",
"metadata": {
"title": "Article Title",
"url": "https://en.wikipedia.org/wiki/Article_Title",
"links": ["Related Article 1", "Related Article 2", ...], # if extract_links=True
"categories": ["Category 1", "Category 2", ...], # if extract_categories=True
"suggestions": ["Suggestion 1", "Suggestion 2", ...], # if article not found
"disambiguation": ["Option 1", "Option 2", ...], # if disambiguation page
"message": "Informational message" # if applicable
},
"format": "wikipedia",
"success": True,
"error": None
}
```
## Metadata Fields
The metadata dictionary may contain different fields depending on the search result:
**Successful Article Found:**
* `title` - The Wikipedia article title
* `url` - Full URL to the Wikipedia article
* `links` - List of related article titles (if `extract_links=True`)
* `categories` - List of article categories (if `extract_categories=True`)
**Disambiguation Page:**
* `title` - The disambiguation page title
* `url` - URL to the disambiguation page
* `disambiguation` - List of possible article titles to choose from
**Article Not Found:**
* `suggestions` - List of suggested article titles that might match
* `message` - Explanation message about why the article wasn't found
## Language Codes
The Wikipedia tool supports searching in any Wikipedia language. Common language codes include:
* `en` - English
* `fr` - French
* `es` - Spanish
* `de` - German
* `it` - Italian
* `pt` - Portuguese
* `ru` - Russian
* `ja` - Japanese
* `zh` - Chinese
* `ar` - Arabic
* `hi` - Hindi
See [Wikipedia Language Editions](https://meta.wikimedia.org/wiki/List_of_Wikipedias) for a complete list.
## Best Practices
1. **Be Specific**: Use specific article titles when possible (e.g., "Python (programming language)" instead of "Python")
2. **Handle Disambiguation**: Always check for disambiguation pages and handle them appropriately
3. **Check Success**: Always verify `result["success"]` before using the data
4. **Limit Summary Length**: Use `max_summary_length` to prevent overly long responses
5. **Extract Only What You Need**: Set `extract_links=False` and `extract_categories=False` if you don't need that data
6. **Cite Sources**: When using with agents, include Wikipedia URLs in responses
7. **Handle Errors Gracefully**: Implement error handling for network issues and missing articles
8. **Use Appropriate Language**: Specify the correct language code for non-English queries
## Performance Considerations
* Wikipedia API requests typically complete in 1-3 seconds
* The tool automatically limits links to 50 and categories to 50 to improve performance
* Use `max_summary_length` to reduce response size when needed
* Only extract links and categories when necessary
## Troubleshooting
### ImportError for requests library
If you encounter "requests library is required" error, install the dependency:
```bash
pip install requests
```
Or install with Peargent:
```bash
pip install peargent[wikipedia]
```
### Network Errors
If you receive network errors, check your internet connection and try again:
```python
result = wikipedia_tool.run({"query": "Python programming"})
if not result["success"]:
print(f"Error: {result['error']}")
# Handle network errors, timeouts, etc.
```
### Invalid Language Code
Ensure you use valid 2-3 letter language codes:
```python
# Valid
result = wikipedia_tool.run({"query": "Topic", "language": "en"})
# Invalid - will return error
result = wikipedia_tool.run({"query": "Topic", "language": "english"})
```
### Article Not Found
When an article isn't found, check the suggestions:
```python
result = wikipedia_tool.run({"query": "Misspelled Topic"})
if not result["text"] and 'suggestions' in result.get('metadata', {}):
print("Did you mean:")
for suggestion in result['metadata']['suggestions']:
print(f" - {suggestion}")
```
### Disambiguation Pages
When you encounter a disambiguation page, choose a more specific query:
```python
result = wikipedia_tool.run({"query": "Python"})
if 'disambiguation' in result.get('metadata', {}):
# Show options to user and search for specific one
result = wikipedia_tool.run({"query": "Python (programming language)"})
```
# Error Handling in Tools
Error Handling in Tools
Comprehensive error handling strategies for tools including retries, timeouts, and validation
Tools can fail for many reasons, network issues, timeouts, invalid API responses, or broken external services. Peargent provides a complete system to handle these failures gracefully.
## Error Handling with `on_error`
The `on_error` parameter controls how tools handle errors (execution failures or validation failures):
```python
from peargent import create_tool
# Option 1: Raise exception (default)
tool_strict = create_tool(
name="critical_api",
description="Critical API call that must succeed",
input_parameters={"query": str},
call_function=call_api,
on_error="raise" # Fail fast if error occurs // [!code highlight]
)
# Option 2: Return error message as string
tool_graceful = create_tool(
name="optional_api",
description="Optional API call",
input_parameters={"query": str},
call_function=call_api,
on_error="return_error" # Continue with error message // [!code highlight]
)
# Option 3: Return None silently
tool_silent = create_tool(
name="analytics_tracker",
description="Optional analytics tracking",
input_parameters={"event": str},
call_function=track_event,
on_error="return_none" # Ignore failures silently // [!code highlight]
)
```
### When to Use Each Strategy
| `on_error` Value | What Happens | What You Can Do | Use Case | Example |
| ----------------------- | ---------------------------------------------------------------------------------------- | -------------------------------------------------------------------- | ------------------------------------------ | -------------------------------------------- |
| **`"raise"`** (default) | Raises exception, stops agent execution | Wrap in `try/except` to catch and handle exception | Critical tools that must succeed | Database writes, payment processing |
| **`"return_error"`** | Returns error message as string (e.g., `"Tool 'api_call' failed: ConnectionError: ..."`) | Check if result is string and contains error, then handle gracefully | Graceful degradation, logging errors | Optional external APIs, analytics |
| **`"return_none"`** | Returns `None` silently, no error message | Check if result is `None`, then use fallback value or skip | Non-critical features, optional enrichment | Analytics tracking, optional data enrichment |
## Next: Advanced Error Handling
Peargent provides more robust failure-handling features:
* **[Retries](/docs/error-handling-in-tools/retries)**\
Automatically retry failing tools with optional exponential backoff.
* **[Timeouts](/docs/error-handling-in-tools/timeout)**\
Prevent long-running or hanging operations.
* **[Validation Failures](/docs/structured-output/tools-output#validating-tool-output-with-schema)**\
Handle schema validation errors when using `output_schema`.
These pages go deeper into reliability patterns for production workloads.
# Retries
Retries
Tools can automatically retry failed operations.
Retries are one of the simplest and most effective ways to handle errors in tools.
Instead of failing immediately, a tool can automatically try again when something goes wrong.
This makes your workflows more resilient, reduces unnecessary crashes, and improves overall reliability with minimal setup.
## Error Handling with Retries
Below is how simple it is to enable retry logic in a tool.
```python
from peargent import create_tool
api_tool = create_tool(
name="external_api",
description="Call external API",
input_parameters={"query": str},
call_function=call_external_api,
max_retries=3, # Retry up to 3 times on failure // [!code highlight:3]
retry_delay=1.0, # Initial delay: 1 second
retry_backoff=True, # Exponential backoff: 1s → 2s → 4s
on_error="return_error"
)
```
## Retry Parameters
| Parameter | Type | Default | Description |
| --------------- | ----- | ------- | ---------------------------------------------------- |
| `max_retries` | int | `0` | Number of retry attempts (0 = no retries) |
| `retry_delay` | float | `1.0` | Initial delay between retries in seconds |
| `retry_backoff` | bool | `True` | Doubles delay after each retry attempt (Exponential) |
## How Retry Works
### First Attempt:
Tool executes normally.
### On Failure:
If execution or validation fails:
* If `max_retries > 0`, the tool waits for retry\_delay seconds
* If `retry_backoff=True`, the wait time doubles each retry
(1s → 2s → 4s → …)
### Repeat:
Retries continue until:
* A retry succeeds, **or**
* All retry attempts are exhausted
### Final Failure:
Handled according to the `on_error` strategy (`raise`, `return_error`, `return_none`).
## Retry Example with Backoff
```python
unreliable_tool = create_tool(
name="flaky_api",
description="API that sometimes fails",
input_parameters={"query": str},
call_function=call_flaky_api,
max_retries=3,
retry_delay=1.0,
retry_backoff=True,
on_error="return_error"
)
# If all attempts fail, timing will be:
# Attempt 1: Immediate
# Attempt 2: +1 second (1.0 * 2^0)
# Attempt 3: +2 seconds (1.0 * 2^1)
# Attempt 4: +4 seconds (1.0 * 2^2)
# Final: Return error message
```
# Timeout
Timeout
Timeouts let you set a maximum allowed execution time for a tool.
If the **Tool** takes longer than the configured time, its execution is stopped and handled using on\_error.
**Timeouts** are extremely useful for:
* Preventing tools from hanging forever
* Stopping slow external operations
* Keeping agent response times predictable
* Automatically failing or retrying long-running tasks
## Enable Timeout in a Tool
```python
from peargent import create_tool
slow_tool = create_tool(
name="slow_operation",
description="Operation that may take too long",
input_parameters={"data": dict},
call_function=slow_processing,
timeout=5.0, # Maximum 5 seconds allowed // [!code highlight]
on_error="return_error"
)
```
## How Timeout Works
1. Tool starts executing normally
2. A timer begins (based on timeout)
3. If execution finishes in time → result is returned
4. If it exceeds the timeout →
* Execution is stopped
* A TimeoutError is raised internally
* Result is handled via on\_error
5. If combined with retries, timeout is applied on every attempt
## Timeout Example with Retries
```python
robust_tool = create_tool(
name="robust_api",
description="API call with timeout + retries",
input_parameters={"query": str},
call_function=call_api,
timeout=10.0, # Max 10 seconds per attempt // [!code highlight]
max_retries=3, # Retry if timed out
retry_delay=2.0,
retry_backoff=True,
on_error="return_error"
)
# Example timing if every attempt times out:
# Attempt 1: 10s timeout
# Wait 2s
# Attempt 2: 10s timeout
# Wait 4s
# Attempt 3: 10s timeout
# Wait 8s
# Attempt 4: 10s timeout
# → Final failure handled by on_error
```
# Custom Storage Backends
Custom Storage Backends
Create custom storage backends for Peargent history.
For production-grade backends with complex requirements, you can create a custom storage backend by subclassing `HistoryStore`. This allows you to persist conversation history in any database or storage system of your choice, such as MongoDB, PostgreSQL, Redis, or even a custom API.
## Subclassing HistoryStore
To create a custom store, you need to implement the abstract methods defined in the `HistoryStore` class. Here is a comprehensive example using MongoDB.
### 1. Initialization and Setup
First, set up your class and initialize the database connection. You should also ensure any necessary indexes are created for performance.
```python
from peargent.storage import HistoryStore, Thread, Message
from typing import Dict, List, Optional, Any
from datetime import datetime
class MongoDBHistoryStore(HistoryStore):
"""Custom MongoDB storage backend."""
def __init__(self, connection_string: str, database: str = "peargent"):
from pymongo import MongoClient
self.client = MongoClient(connection_string)
self.db = self.client[database]
self.threads = self.db.threads
self.messages = self.db.messages
# Create indexes for performance
# Indexing 'id' ensures fast thread lookups
self.threads.create_index("id", unique=True)
# Compound index on 'thread_id' and 'timestamp' speeds up message retrieval
self.messages.create_index([("thread_id", 1), ("timestamp", 1)])
```
### 2. Thread Management
Implement methods to create, retrieve, and list threads.
**Creating Threads:**
When creating a thread, you must persist its ID, creation time, and any initial metadata.
```python
def create_thread(self, metadata: Optional[Dict] = None) -> str:
thread = Thread(metadata=metadata)
self.threads.insert_one({
"id": thread.id,
"created_at": thread.created_at,
"updated_at": thread.updated_at,
"metadata": thread.metadata
})
return thread.id
```
**Retrieving Threads:**
When retrieving a thread, you need to reconstruct the `Thread` object from your database record. Crucially, you must also load the associated messages and attach them to the thread.
```python
def get_thread(self, thread_id: str) -> Optional[Thread]:
thread_data = self.threads.find_one({"id": thread_id})
if not thread_data:
return None
thread = Thread(
thread_id=thread_data["id"],
metadata=thread_data.get("metadata", {}),
created_at=thread_data["created_at"],
updated_at=thread_data["updated_at"]
)
# Load messages associated with this thread, sorted by timestamp
messages = self.messages.find({"thread_id": thread_id}).sort("timestamp", 1)
for msg_data in messages:
msg = Message(
role=msg_data["role"],
content=msg_data["content"],
agent=msg_data.get("agent"),
tool_call=msg_data.get("tool_call"),
metadata=msg_data.get("metadata", {}),
message_id=msg_data["id"],
timestamp=msg_data["timestamp"]
)
thread.messages.append(msg)
return thread
```
### 3. Message Persistence
Implement the logic to save new messages.
**Appending Messages:**
This method is called whenever a new message is added to the history. You should save the message and update the thread's `updated_at` timestamp.
```python
def append_message(
self,
thread_id: str,
role: str,
content: Any,
agent: Optional[str] = None,
tool_call: Optional[Dict] = None,
metadata: Optional[Dict] = None
) -> Message:
message = Message(
role=role,
content=content,
agent=agent,
tool_call=tool_call,
metadata=metadata
)
self.messages.insert_one({
"id": message.id,
"thread_id": thread_id,
"timestamp": message.timestamp,
"role": message.role,
"content": message.content,
"agent": message.agent,
"tool_call": message.tool_call,
"metadata": message.metadata
})
# Update thread's updated_at timestamp
self.threads.update_one(
{"id": thread_id},
{"$set": {"updated_at": datetime.now()}}
)
return message
```
### 4. Utility Methods
Implement the remaining utility methods for listing and deleting.
```python
def get_messages(self, thread_id: str) -> List[Message]:
"""Retrieve all messages for a specific thread."""
thread = self.get_thread(thread_id)
return thread.messages if thread else []
def list_threads(self) -> List[str]:
"""Return a list of all thread IDs."""
return [t["id"] for t in self.threads.find({}, {"id": 1})]
def delete_thread(self, thread_id: str) -> bool:
"""Delete a thread and all its associated messages."""
result = self.threads.delete_one({"id": thread_id})
if result.deleted_count > 0:
self.messages.delete_many({"thread_id": thread_id})
return True
return False
```
### Usage
Once your class is defined, you can use it just like any built-in storage backend.
#### With create\_agent (Automatic Integration)
You can pass your custom storage backend directly to `create_agent` using `HistoryConfig`:
```python
from peargent import create_agent, HistoryConfig
from peargent.models import openai
# Initialize your custom store
store = MongoDBHistoryStore(connection_string="mongodb://localhost:27017")
# Create agent with custom storage backend
agent = create_agent(
name="Assistant",
description="A helpful assistant with MongoDB history",
persona="You are a helpful AI assistant.",
model=openai("gpt-4o"),
history=HistoryConfig(
auto_manage_context=True,
max_context_messages=20,
strategy="smart",
store=store # Your custom storage backend
)
)
# Use the agent - history is automatically managed
response1 = agent.run("My name is Alice")
# Agent creates thread and stores message in MongoDB
response2 = agent.run("What's my name?")
# Agent loads history from MongoDB and remembers: "Your name is Alice"
```
#### With create\_pool (Multi-Agent with Custom Storage)
You can also use custom storage backends with agent pools for shared history across multiple agents:
```python
from peargent import create_agent, create_pool, HistoryConfig
from peargent.models import openai
# Initialize your custom store
store = MongoDBHistoryStore(connection_string="mongodb://localhost:27017")
# Create multiple agents
researcher = create_agent(
name="Researcher",
description="Researches topics thoroughly",
persona="You are a detail-oriented researcher.",
model=openai("gpt-4o-mini")
)
writer = create_agent(
name="Writer",
description="Writes clear summaries",
persona="You are a skilled technical writer.",
model=openai("gpt-4o")
)
# Create pool with custom storage - all agents share the same MongoDB history
pool = create_pool(
agents=[researcher, writer],
default_model=openai("gpt-4o"),
history=HistoryConfig(
auto_manage_context=True,
max_context_messages=25,
strategy="smart",
store=store # Shared custom storage for all agents
)
)
# Use the pool
result = pool.run("Research quantum computing and write a summary")
# Both agents' interactions are stored in MongoDB
```
# History Management
History Management
Custom storage backends, manual thread control, and low-level history operations for advanced use cases.
This guide covers advanced history capabilities for developers who need fine-grained control over conversation persistence, custom storage implementations, and low-level message operations.
## Next Steps
* See **[History](/docs/history)** for basic usage and auto-management
* See **[Agents](/docs/agents)** for integrating history with agents
* See **[Pools](/docs/pools)** for shared history across multiple agents
# Manual Thread Management
Manual Thread Management
Manually control threads for multi-user applications.
While agents automatically manage threads, you can also control threads manually for multi-user applications or complex workflows.
## Creating and Switching Threads
You can create multiple threads to handle different conversations simultaneously. By using `create_thread` with metadata, you can tag threads with user IDs or session info. The `use_thread` method then lets you switch the active context, ensuring that new messages go to the correct conversation.
```python
from peargent import create_history
from peargent.storage import Sqlite
history = create_history(store_type=Sqlite(database_path="./app.db"))
# Create threads for different users
alice_thread = history.create_thread(metadata={"user_id": "alice", "session": "web"})
bob_thread = history.create_thread(metadata={"user_id": "bob", "session": "mobile"})
# Switch between threads
history.use_thread(alice_thread)
history.add_user_message("What's the weather?")
history.use_thread(bob_thread)
history.add_user_message("Show my orders")
# List all threads
all_threads = history.list_threads()
print(f"Total threads: {len(all_threads)}")
# Get thread with metadata
thread = history.get_thread(alice_thread)
print(f"User: {thread.metadata.get('user_id')}")
print(f"Messages: {len(thread.messages)}")
```
## Multi-User Application Pattern
For applications serving multiple users, you need to ensure each user gets their own conversation history. This pattern shows how to use metadata to look up an existing thread for a user. If a thread is found, the agent resumes that conversation; if not, a new thread is created. This allows a single agent instance to handle many users concurrently.
```python
from peargent import create_agent, create_history, HistoryConfig
from peargent.storage import Postgresql
from peargent.models import openai
# Shared history store for all users
history = create_history(
store_type=Postgresql(
connection_string="postgresql://user:pass@localhost/app_db"
)
)
# Create agent (reused across users)
agent = create_agent(
name="Assistant",
description="Customer support assistant",
persona="You are a helpful customer support agent.",
model=openai("gpt-4o")
)
def handle_user_message(user_id: str, message: str):
"""Handle message from a specific user."""
# Find or create thread for this user
all_threads = history.list_threads()
user_thread = None
for thread_id in all_threads:
thread = history.get_thread(thread_id)
if thread.metadata.get("user_id") == user_id:
user_thread = thread_id
break
if not user_thread:
# Create new thread for this user
user_thread = history.create_thread(metadata={"user_id": user_id})
# Set active thread
history.use_thread(user_thread)
# Add user message
history.add_user_message(message)
# Get response from agent
# Note: Agent needs to load this history manually or use temporary_memory
response = agent.run(message)
# Add assistant response
history.add_assistant_message(response, agent="Assistant")
return response
# Usage
response1 = handle_user_message("alice", "What's my order status?")
response2 = handle_user_message("bob", "I need help with returns")
response3 = handle_user_message("alice", "Thanks!") # Same thread as first message
```
# History API Reference
History API Reference
Complete reference for Thread, Message, and Context operations.
## Thread Operations
### create\_thread
Creates a new conversation thread and sets it as the active thread.
```python
thread_id = history.create_thread(metadata={
"user_id": "alice",
"topic": "customer_support",
"tags": ["billing", "urgent"]
})
```
**Parameters**
| Name | Type | Default | Description |
| :--------- | :--------------- | :------ | :----------------------------------------------------- |
| `metadata` | `Optional[Dict]` | `None` | Dictionary of custom metadata to attach to the thread. |
**Returns**
* `str`: The unique ID of the created thread.
***
### use\_thread
Switches the active context to an existing thread.
```python
history.use_thread("thread-123-abc")
```
**Parameters**
| Name | Type | Default | Description |
| :---------- | :---- | :------ | :--------------------------------- |
| `thread_id` | `str` | - | The ID of the thread to switch to. |
***
### get\_thread
Retrieves a thread object.
```python
thread = history.get_thread()
print(f"Created: {thread.created_at}")
```
**Parameters**
| Name | Type | Default | Description |
| :---------- | :-------------- | :------ | :------------------------------------------------------------------------------------ |
| `thread_id` | `Optional[str]` | `None` | The ID of the thread to retrieve. If not provided, returns the current active thread. |
**Returns**
* `Optional[Thread]`: The thread object, or `None` if not found.
***
### list\_threads
Lists all available thread IDs in the storage.
```python
all_threads = history.list_threads()
print(f"Total threads: {len(all_threads)}")
```
**Returns**
* `List[str]`: A list of all thread IDs.
***
### delete\_thread
Deletes a thread and all its associated messages.
```python
if history.delete_thread(thread_id):
print("Thread deleted")
```
**Parameters**
| Name | Type | Default | Description |
| :---------- | :---- | :------ | :------------------------------ |
| `thread_id` | `str` | - | The ID of the thread to delete. |
**Returns**
* `bool`: `True` if the thread was successfully deleted, `False` otherwise.
## Message Operations
### add\_user\_message
Adds a user message to the current thread.
```python
msg = history.add_user_message(
"What's the weather today?",
metadata={"source": "web"}
)
```
**Parameters**
| Name | Type | Default | Description |
| :--------- | :--------------- | :------ | :------------------------------- |
| `content` | `str` | - | The text content of the message. |
| `metadata` | `Optional[Dict]` | `None` | Custom metadata for the message. |
**Returns**
* `Message`: The created message object.
***
### add\_assistant\_message
Adds an assistant response to the current thread.
```python
msg = history.add_assistant_message(
"The weather is sunny.",
agent="WeatherBot",
metadata={"model": "gpt-4o"}
)
```
**Parameters**
| Name | Type | Default | Description |
| :--------- | :--------------- | :------ | :-------------------------------------------------- |
| `content` | `Any` | - | The content of the response (string or structured). |
| `agent` | `Optional[str]` | `None` | Name of the agent that generated the response. |
| `metadata` | `Optional[Dict]` | `None` | Custom metadata (e.g., tokens used, model name). |
**Returns**
* `Message`: The created message object.
***
### add\_tool\_message
Adds a tool execution result to the current thread.
```python
msg = history.add_tool_message(
tool_call={
"name": "get_weather",
"output": {"temp": 72}
},
agent="WeatherBot"
)
```
**Parameters**
| Name | Type | Default | Description |
| :---------- | :--------------- | :------ | :----------------------------------------------------------------- |
| `tool_call` | `Dict` | - | Dictionary containing tool execution details (name, args, output). |
| `agent` | `Optional[str]` | `None` | Name of the agent that called the tool. |
| `metadata` | `Optional[Dict]` | `None` | Custom metadata (e.g., execution time). |
**Returns**
* `Message`: The created message object.
***
### get\_messages
Retrieves messages from a thread with optional filtering.
```python
# Get only user messages
user_messages = history.get_messages(role="user")
```
**Parameters**
| Name | Type | Default | Description |
| :---------- | :-------------- | :------ | :------------------------------------------------------ |
| `thread_id` | `Optional[str]` | `None` | Target thread ID. Defaults to current thread. |
| `role` | `Optional[str]` | `None` | Filter by role ("user", "assistant", "tool", "system"). |
| `agent` | `Optional[str]` | `None` | Filter by agent name. |
**Returns**
* `List[Message]`: List of matching message objects.
***
### get\_message\_count
Gets the total number of messages in a thread.
```python
count = history.get_message_count()
```
**Parameters**
| Name | Type | Default | Description |
| :---------- | :-------------- | :------ | :-------------------------------------------- |
| `thread_id` | `Optional[str]` | `None` | Target thread ID. Defaults to current thread. |
**Returns**
* `int`: The number of messages.
***
### delete\_message
Deletes a specific message by its ID.
```python
history.delete_message(message_id)
```
**Parameters**
| Name | Type | Default | Description |
| :----------- | :-------------- | :------ | :-------------------------------------------- |
| `message_id` | `str` | - | The ID of the message to delete. |
| `thread_id` | `Optional[str]` | `None` | Target thread ID. Defaults to current thread. |
**Returns**
* `bool`: `True` if deleted successfully.
***
### delete\_messages
Deletes multiple messages at once.
```python
history.delete_messages([msg_id_1, msg_id_2])
```
**Parameters**
| Name | Type | Default | Description |
| :------------ | :-------------- | :------ | :-------------------------------------------- |
| `message_ids` | `List[str]` | - | List of message IDs to delete. |
| `thread_id` | `Optional[str]` | `None` | Target thread ID. Defaults to current thread. |
**Returns**
* `int`: The number of messages actually deleted.
## Context Management Operations
### trim\_messages
Trims messages to manage context window size.
```python
# Keep only the last 10 messages
history.trim_messages(strategy="last", count=10)
```
**Parameters**
| Name | Type | Default | Description |
| :------------ | :-------------- | :------- | :------------------------------------------------------------------------- |
| `strategy` | `str` | `"last"` | Strategy: `"last"` (keep recent), `"first"` (keep oldest), `"first_last"`. |
| `count` | `int` | `10` | Number of messages to keep. |
| `keep_system` | `bool` | `True` | If `True`, system messages are never deleted. |
| `thread_id` | `Optional[str]` | `None` | Target thread ID. |
**Returns**
* `int`: Number of messages removed.
***
### summarize\_messages
Summarizes a range of messages using an LLM and replaces them with a summary.
```python
history.summarize_messages(
model=groq("llama-3.1-8b-instant"),
keep_recent=5
)
```
**Parameters**
| Name | Type | Default | Description |
| :------------ | :-------------- | :------ | :------------------------------------------------- |
| `model` | `Any` | - | LLM model instance for generating summary. |
| `start_index` | `int` | `0` | Start index for summarization. |
| `end_index` | `Optional[int]` | `None` | End index (defaults to `len - keep_recent`). |
| `keep_recent` | `int` | `5` | Number of recent messages to exclude from summary. |
| `thread_id` | `Optional[str]` | `None` | Target thread ID. |
**Returns**
* `Message`: The newly created summary message.
***
### manage\_context\_window
Automatically manages context window when messages exceed a threshold.
```python
history.manage_context_window(
model=groq("llama-3.1-8b-instant"),
max_messages=20,
strategy="smart"
)
```
**Parameters**
| Name | Type | Default | Description |
| :------------- | :-------------- | :-------- | :----------------------------------------------------------------- |
| `model` | `Any` | - | LLM model (required for "summarize" and "smart"). |
| `max_messages` | `int` | `20` | Threshold to trigger management. |
| `strategy` | `str` | `"smart"` | Strategy: `"smart"`, `"trim_last"`, `"trim_first"`, `"summarize"`. |
| `thread_id` | `Optional[str]` | `None` | Target thread ID. |
## Data Models
### Message Object
Represents a single message in the conversation history.
| Property | Type | Description |
| :---------- | :--------------- | :---------------------------------------------------------- |
| `id` | `str` | Unique UUID for the message. |
| `timestamp` | `datetime` | Time when the message was created. |
| `role` | `str` | Role of the sender (`user`, `assistant`, `tool`, `system`). |
| `content` | `Any` | The content of the message. |
| `agent` | `Optional[str]` | Name of the agent (for assistant messages). |
| `tool_call` | `Optional[Dict]` | Tool execution details (for tool messages). |
| `metadata` | `Dict` | Custom metadata dictionary. |
### Thread Object
Represents a conversation thread containing multiple messages.
| Property | Type | Description |
| :----------- | :-------------- | :-------------------------------------- |
| `id` | `str` | Unique UUID for the thread. |
| `created_at` | `datetime` | Time when the thread was created. |
| `updated_at` | `datetime` | Time when the thread was last modified. |
| `metadata` | `Dict` | Custom metadata dictionary. |
| `messages` | `List[Message]` | List of messages in the thread. |
# Optimizing for Cost
Optimizing for Cost
Strategies to control token usage and reduce API costs
Running LLM agents in production can be expensive if not managed carefully. This guide outlines practical strategies to keep your costs under control without sacrificing the quality of your agent's responses.
## 1. Choose the Right Model
Not every task requires the most powerful model. Match your model choice to the task complexity.
### Model Selection Strategy
```python
from peargent import create_agent
from peargent.models import groq
# Use smaller models for simple tasks
classifier_agent = create_agent(
name="Classifier",
description="Classifies user intent",
persona="You classify user messages into categories: support, sales, or general.",
model=groq("llama-3.1-8b") # [!code highlight] - Smaller, cheaper model
)
# Use larger models only for complex tasks
reasoning_agent = create_agent(
name="Reasoner",
description="Solves complex problems",
persona="You solve complex reasoning and coding problems step by step.",
model=groq("llama-3.3-70b-versatile") # [!code highlight] - Larger model when needed
)
```
**Guidelines:**
* **Simple tasks** (classification, extraction, summarization): Use smaller models (8B parameters)
* **Complex tasks** (reasoning, coding, analysis): Use larger models (70B+ parameters)
* **Test different models** on your specific use case to find the best cost/quality balance
### Track Model Costs with Custom Pricing
If you're using custom or local models, add their pricing to track costs accurately:
```python
from peargent.observability import enable_tracing, get_tracer
tracer = enable_tracing()
# Add custom pricing for your model (prices per million tokens)
tracer.add_custom_pricing( # [!code highlight]
model="my-fine-tuned-model",
prompt_price=1.50, # $1.50 per million prompt tokens
completion_price=3.00 # $3.00 per million completion tokens
)
# Now cost tracking works for your custom model
agent = create_agent(
name="CustomAgent",
model=my_custom_model,
persona="You are helpful.",
tracing=True
)
```
## 2. Control Context with History Management
The context window is your biggest cost driver. Every message in the conversation history is re-sent with each request.
### Configure Automatic Context Management
Use `HistoryConfig` to automatically manage conversation history:
```python
from peargent import create_agent, HistoryConfig
from peargent.storage import InMemory
from peargent.models import groq
agent = create_agent(
name="CostOptimizedAgent",
description="Agent with automatic history management",
persona="You are a helpful assistant.",
model=groq("llama-3.3-70b-versatile"),
history=HistoryConfig( # [!code highlight]
auto_manage_context=True, # Enable automatic management
max_context_messages=10, # Keep only last 10 messages
strategy="trim_last", # Remove oldest messages when limit reached
store=InMemory()
)
)
```
### Available Context Strategies
Peargent supports 5 context management strategies:
| Strategy | How It Works | Use When | Cost Impact |
| -------------- | ----------------------------------- | --------------------------- | ----------------------------- |
| `"trim_last"` | Removes oldest messages | Simple conversations | ✅ Low - fast, no LLM calls |
| `"trim_first"` | Keeps oldest messages | Important initial context | ✅ Low - fast, no LLM calls |
| `"first_last"` | Keeps first and last messages | Preserving original context | ✅ Low - fast, no LLM calls |
| `"summarize"` | Summarizes old messages | Complex conversations | ⚠️ Medium - requires LLM call |
| `"smart"` | Chooses best strategy automatically | General purpose | ⚠️ Variable - may use LLM |
**Example: Trim Strategy (Recommended for Cost)**
```python
# Most cost-effective - no LLM calls for management
history=HistoryConfig(
auto_manage_context=True,
max_context_messages=10, # [!code highlight] - Only keep 10 messages
strategy="trim_last", # [!code highlight] - Drop oldest messages
store=InMemory()
)
```
**Example: Summarize Strategy (Better Context Retention)**
```python
# Uses LLM to summarize old messages - costs more but retains context
history=HistoryConfig(
auto_manage_context=True,
max_context_messages=20,
strategy="summarize", # [!code highlight] - Summarize old messages
summarize_model=groq("llama-3.1-8b"), # [!code highlight] - Use cheap model for summaries
store=InMemory()
)
```
**Example: Smart Strategy (Balanced)**
```python
# Automatically chooses between trim and summarize
history=HistoryConfig(
auto_manage_context=True,
max_context_messages=15,
strategy="smart", # [!code highlight] - Automatically adapts
store=InMemory()
)
```
## 3. Limit Output Length with max\_tokens
Control how much the agent can generate by setting `max_tokens` in model parameters:
```python
from peargent import create_agent
from peargent.models import groq
# Limit output to reduce costs
agent = create_agent(
name="BriefAgent",
description="Gives brief responses",
persona="You provide concise, brief answers. Maximum 2-3 sentences.",
model=groq(
"llama-3.3-70b-versatile",
parameters={
"max_tokens": 150, # [!code highlight] - Limit to ~150 tokens output
"temperature": 0.7
}
)
)
response = agent.run("Explain quantum computing")
# Agent cannot generate more than 150 tokens
```
**Guidelines:**
* Short answers: `max_tokens=150` (\~100 words)
* Medium answers: `max_tokens=500` (\~350 words)
* Long answers: `max_tokens=2000` (\~1500 words)
* Code generation: `max_tokens=4096` or higher
### Move Examples to Tool Descriptions
Instead of putting examples in the persona, put them in tool descriptions:
```python
from peargent import create_agent, create_tool
def search_database(query: str) -> str:
# Implementation...
return "Results found"
agent = create_agent(
name="ProductAgent",
persona="You help with product inquiries.", # Short persona
model=groq("llama-3.3-70b-versatile"),
tools=[create_tool(
name="search_database",
description="""Searches the product database for matching items.
Use this tool when users ask about products, inventory, or availability.
Examples: "Do we have red shirts?" → use this tool with query="red shirts"
"Check stock for item #123" → use this tool with query="item 123"
""", # [!code highlight] - Examples in tool description, not persona
input_parameters={"query": str},
call_function=search_database
)]
)
```
## 4. Control Temperature for Deterministic Outputs
Lower temperature reduces token usage for tasks that need deterministic outputs:
```python
from peargent import create_agent
from peargent.models import groq
# For deterministic tasks (extraction, classification)
extraction_agent = create_agent(
name="Extractor",
description="Extracts structured data",
persona="Extract the requested information exactly as it appears.",
model=groq(
"llama-3.3-70b-versatile",
parameters={
"temperature": 0.0, # [!code highlight] - Deterministic, shorter outputs
"max_tokens": 500
}
)
)
# For creative tasks (writing, brainstorming)
creative_agent = create_agent(
name="Writer",
description="Writes creative content",
persona="You write engaging, creative content.",
model=groq(
"llama-3.3-70b-versatile",
parameters={
"temperature": 0.9, # [!code highlight] - More creative, longer outputs
"max_tokens": 2000
}
)
)
```
## 5. Monitor Costs with Tracing
You can't optimize what you can't measure. Use Peargent's observability features to track costs.
### Enable Cost Tracking
```python
from peargent import create_agent
from peargent.observability import enable_tracing
from peargent.storage import Sqlite
from peargent.models import groq
# Enable tracing with database storage
tracer = enable_tracing(
store_type=Sqlite(connection_string="sqlite:///./traces.db")
)
agent = create_agent(
name="TrackedAgent",
description="Agent with cost tracking",
persona="You are helpful.",
model=groq("llama-3.3-70b-versatile"),
tracing=True # [!code highlight] - Enable tracing for this agent
)
# Use the agent
response = agent.run("Hello")
# Check costs
traces = tracer.list_traces()
latest = traces[-1]
print(f"Cost: ${latest.total_cost:.6f}")
print(f"Tokens: {latest.total_tokens}")
print(f"Duration: {latest.duration_ms}ms")
```
### Analyze Cost Patterns
```python
from peargent.observability import get_tracer
tracer = get_tracer()
# Get aggregate statistics
stats = tracer.get_aggregate_stats() # [!code highlight]
print(f"Total Traces: {stats['total_traces']}")
print(f"Total Cost: ${stats['total_cost']:.6f}")
print(f"Average Cost per Trace: ${stats['avg_cost_per_trace']:.6f}")
print(f"Total Tokens: {stats['total_tokens']:,}")
# Find expensive operations
traces = tracer.list_traces()
expensive_traces = sorted(traces, key=lambda t: t.total_cost, reverse=True)[:5]
print("\nMost Expensive Operations:")
for trace in expensive_traces:
print(f" {trace.agent_name}: ${trace.total_cost:.6f} ({trace.total_tokens} tokens)")
```
### Set Cost Alerts
```python
from peargent.observability import get_tracer
tracer = get_tracer()
MAX_COST_PER_REQUEST = 0.01 # $0.01 limit
for update in agent.stream_observe(user_input):
if update.is_agent_end:
if update.cost > MAX_COST_PER_REQUEST: # [!code highlight]
print(f"⚠️ WARNING: Cost ${update.cost:.6f} exceeds limit!")
# Log alert, notify admins, etc.
```
### Track Costs by User
```python
from peargent.observability import enable_tracing, set_user_id, get_tracer
from peargent.storage import Postgresql
tracer = enable_tracing(
store_type=Postgresql(connection_string="postgresql://user:pass@localhost/db")
)
# Set user ID before agent runs
set_user_id("user_123") # [!code highlight]
agent.run("Hello")
# Get costs for specific user
user_stats = tracer.get_aggregate_stats(user_id="user_123") # [!code highlight]
print(f"User 123 total cost: ${user_stats['total_cost']:.6f}")
```
## 6. Use Streaming to Show Progress
While streaming doesn't reduce costs, it improves perceived performance, making slower/cheaper models feel faster:
```python
from peargent import create_agent
from peargent.models import groq
# Use cheaper model with streaming
agent = create_agent(
name="StreamingAgent",
description="Shows progress immediately",
persona="You are helpful.",
model=groq("llama-3.1-8b") # Cheaper model
)
# Stream response - user sees first token in ~200ms
print("Agent: ", end="", flush=True)
for chunk in agent.stream("Explain AI"): # [!code highlight]
print(chunk, end="", flush=True)
```
**Benefit:** Cheaper models feel faster with streaming, reducing pressure to use expensive models.
## 7. Count Tokens Before Sending
Estimate costs before making expensive calls:
```python
from peargent.observability import get_cost_tracker
tracker = get_cost_tracker()
# Count tokens in your prompt
prompt = "Explain quantum computing in detail..."
token_count = tracker.count_tokens(prompt, model="llama-3.3-70b-versatile") # [!code highlight]
print(f"Prompt will use ~{token_count} tokens")
# Estimate cost
estimated_cost = tracker.calculate_cost( # [!code highlight]
prompt_tokens=token_count,
completion_tokens=500, # Estimate 500 token response
model="llama-3.3-70b-versatile"
)
print(f"Estimated cost: ${estimated_cost:.6f}")
# Decide whether to proceed
if estimated_cost > 0.01:
print("Too expensive! Shortening prompt...")
# Truncate or summarize prompt
```
## Cost Optimization Checklist
Use this checklist for production deployments:
### Model Selection
* [ ] Using smallest viable model for each agent type
* [ ] Tested cost vs quality tradeoff for your use case
* [ ] Custom pricing configured for local/fine-tuned models
### Context Management
* [ ] `HistoryConfig` configured with appropriate strategy
* [ ] `max_context_messages` set to reasonable limit (10-20)
* [ ] Using "trim\_last" for cost-sensitive applications
* [ ] Cheaper model used for summarization if using "summarize" strategy
### Output Control
* [ ] `max_tokens` set based on expected response length
* [ ] Persona/system prompt optimized for brevity
* [ ] Examples moved from persona to tool descriptions
* [ ] Temperature set to 0.0 for deterministic tasks
### Monitoring
* [ ] Tracing enabled in production
* [ ] Cost tracking configured with accurate pricing
* [ ] Regular analysis of aggregate statistics
* [ ] Alerts set for expensive operations
* [ ] Per-user cost tracking implemented
### Implementation
* [ ] Token counting used for cost estimation
* [ ] Streaming enabled for better UX with cheaper models
* [ ] Cost limits enforced in application logic
* [ ] Regular review of most expensive operations
## Summary
**Biggest Cost Savings:**
1. **History Management** - Use `trim_last` with `max_context_messages=10` (saves 50-80% on tokens)
2. **Model Selection** - Use smaller models for simple tasks (saves 50-90% on costs)
3. **Persona Optimization** - Short personas (saves 5-10% per request)
4. **max\_tokens** - Limit output length (saves 20-40% on completion tokens)
**Essential Monitoring:**
* Enable tracing in production
* Track costs per user/session
* Analyze aggregate statistics weekly
* Set alerts for expensive operations
Start with history management and model selection for the biggest impact!
# Creating Custom Tooling
Creating Custom Tooling
Best practices for building robust, reusable tools for your agents
Tools are the hands of your agent—they allow it to interact with the outside world, from searching the web to querying databases. While Peargent comes with built-in tools, the real power lies in creating custom tools tailored to your specific needs.
## The Anatomy of a Tool
Every tool in Peargent has four essential components:
```python
from peargent import create_tool
def search_database(query: str) -> str:
# Your implementation here
return "Results found"
tool = create_tool(
name="search_database", # [!code highlight] - Tool identifier
description="""Searches the product database for matching items.
Use this tool when users ask about products, inventory, or availability.""", # [!code highlight] - What LLM sees
input_parameters={"query": str}, # [!code highlight] - Expected arguments
call_function=search_database # [!code highlight] - Function to execute
)
```
**The Four Components:**
1. **name**: Unique identifier the LLM uses to call the tool
2. **description**: Tells the LLM *what* the tool does and *when* to use it
3. **input\_parameters**: Dict mapping parameter names to their types
4. **call\_function**: The Python function that implements the tool logic
## The Golden Rules of Tool Building
### 1. Descriptive Descriptions are Mandatory
The LLM uses the tool's `description` parameter to understand *what* the tool does and *when* to use it. Be verbose and precise.
#### Bad: Vague Description
```python
tool = create_tool(
name="fetch_user",
description="Gets user info.", # ❌ Too vague!
input_parameters={"user_id": str},
call_function=fetch_user
)
```
#### Good: Clear and Specific
```python
tool = create_tool(
name="fetch_user_data",
description="""Retrieves detailed profile information for a specific user from the database.
Use this tool when you need to look up:
- User's email address
- Phone number
- Account status (active/suspended)
- Registration date
Do NOT use this for:
- Searching users by name (use search_users instead)
- Updating user data (use update_user instead)
Returns: Dict with keys: user_id, email, phone, status, registered_at""", # ✅ Clear when to use it!
input_parameters={"user_id": str},
call_function=fetch_user_data
)
```
**Why This Matters:** The LLM decides which tool to call based solely on the description. A clear description = correct tool selection.
### 2. Type Hinting is Critical
Peargent uses type hints to generate the JSON schema for the LLM. Always type your arguments and return values.
#### Bad: No Type Hints
```python
def calculate_tax(amount, region): # ❌ LLM can't infer types!
return amount * 0.1
```
#### Good: Full Type Hints
```python
def calculate_tax(amount: float, region: str) -> float: # ✅ Clear types!
"""Calculates tax based on amount and region."""
tax_rates = {"CA": 0.0725, "NY": 0.08875, "TX": 0.0625}
return amount * tax_rates.get(region, 0.05)
```
**Supported Types:**
* Primitives: `str`, `int`, `float`, `bool`
* Collections: `list`, `dict`, `list[str]`, `dict[str, int]`
* Pydantic Models: For complex structured inputs (see below)
### 3. Handle Errors Gracefully
Tools should not crash the agent. Use the `on_error` parameter to control failure behavior.
```python
from peargent import create_tool
def search_database(query: str) -> str:
"""Searches the database for results."""
try:
results = db.execute(query)
return str(results)
except Exception as e:
return f"Error executing query: {str(e)}. Please check your syntax."
# Three error handling strategies:
# Strategy 1: RAISE (Default) - Strict mode
critical_tool = create_tool(
name="critical_operation",
description="Must succeed - handles payment",
input_parameters={"amount": float},
call_function=process_payment,
on_error="raise" # [!code highlight] - Crash if fails (default)
)
# Strategy 2: RETURN_ERROR - Graceful mode
optional_tool = create_tool(
name="get_recommendations",
description="Optional product recommendations",
input_parameters={"user_id": str},
call_function=get_recommendations,
on_error="return_error" # [!code highlight] - Return error message as string
)
# Strategy 3: RETURN_NONE - Silent mode
analytics_tool = create_tool(
name="log_event",
description="Logs analytics events",
input_parameters={"event": str},
call_function=log_analytics,
on_error="return_none" # [!code highlight] - Return None silently
)
```
**When to Use Each Strategy:**
| Strategy | Use Case | Example |
| ---------------- | ------------------------------------------- | ------------------------------------------------ |
| `"raise"` | Critical operations that must succeed | Authentication, payments, database writes |
| `"return_error"` | Optional features that shouldn't break flow | Recommendations, third-party APIs, cache lookups |
| `"return_none"` | Nice-to-have features | Analytics, logging, notifications |
### 4. Keep It Simple (Idempotency)
Ideally, tools should be **idempotent**—calling them multiple times with the same arguments should produce the same result. Avoid tools that rely heavily on hidden state.
#### Bad: Stateful Tool
```python
counter = 0 # ❌ Hidden state!
def increment() -> int:
"""Increments counter."""
global counter
counter += 1
return counter
```
#### Good: Stateless Tool
```python
def get_user_count() -> int:
"""Gets current user count from database."""
return db.query("SELECT COUNT(*) FROM users").scalar() # ✅ Same input = same output
```
## Advanced Features
### Complex Input with Pydantic
For tools with many parameters or nested data, use Pydantic models for input.
```python
from pydantic import BaseModel, Field
from peargent import create_tool
class TicketInput(BaseModel):
title: str = Field(..., description="Brief summary of the issue")
priority: str = Field(..., description="Ticket priority")
description: str = Field(..., description="Detailed description")
category: str = Field(default="general", description="Ticket category")
# Validation
def __init__(self, **data):
# Validate priority
if data.get("priority") not in ["LOW", "MEDIUM", "HIGH"]:
raise ValueError("Priority must be LOW, MEDIUM, or HIGH")
super().__init__(**data)
def create_support_ticket(data: TicketInput) -> str:
"""
Creates a new support ticket in the system.
Required fields:
- title: Brief summary (e.g., "Cannot login")
- priority: LOW, MEDIUM, or HIGH
- description: Detailed explanation
Optional fields:
- category: Ticket category (default: "general")
"""
# Access validated fields
ticket_id = db.create_ticket(
title=data.title,
priority=data.priority,
description=data.description,
category=data.category
)
return f"Ticket #{ticket_id} created with priority {data.priority}"
ticket_tool = create_tool(
name="create_ticket",
description="""Creates a new support ticket in the system.
Required fields:
- title: Brief summary (e.g., "Cannot login")
- priority: LOW, MEDIUM, or HIGH
- description: Detailed explanation
Optional fields:
- category: Ticket category (default: "general")""",
input_parameters={"data": TicketInput}, # [!code highlight] - Use Pydantic model
call_function=create_support_ticket
)
```
**Benefits:**
* Clear parameter structure
* Built-in validation
* Default values
* Nested objects supported
## Real-World Examples
### Example 1: Database Query Tool
```python
from peargent import create_tool
import sqlite3
def query_products(
category: str,
min_price: float = 0.0,
max_price: float = 999999.99
) -> str:
"""
Searches the product database by category and price range.
Use this tool when users ask about:
- Products in a specific category
- Products within a price range
- Available inventory
Examples:
- "Show me electronics under $500" → category="electronics", max_price=500
- "What furniture do you have?" → category="furniture"
Returns: Formatted list of matching products with prices
"""
try:
conn = sqlite3.connect("products.db")
cursor = conn.cursor()
query = """
SELECT name, price, stock
FROM products
WHERE category = ? AND price BETWEEN ? AND ?
ORDER BY price
"""
results = cursor.fetchall()
conn.close()
if not results:
return f"No products found in '{category}' between ${min_price} and ${max_price}"
# Format results
output = f"Found {len(results)} products in '{category}':\n\n"
for name, price, stock in results:
output += f"- {name}: ${price:.2f} ({stock} in stock)\n"
return output
except Exception as e:
return f"Database error: {str(e)}"
product_tool = create_tool(
name="query_products",
description="""Searches the product database by category and price range.
Use this tool when users ask about:
- Products in a specific category
- Products within a price range
- Available inventory
Examples:
- "Show me electronics under $500" → category="electronics", max_price=500
- "What furniture do you have?" → category="furniture"
Returns: Formatted list of matching products with prices""",
input_parameters={
"category": str,
"min_price": float,
"max_price": float
},
call_function=query_products,
on_error="return_error" # Don't crash on DB errors
)
```
### Example 2: External API
```python
from peargent import create_tool
import requests
def fetch_weather(city: str) -> str:
"""
Fetches current weather data from OpenWeatherMap API.
Use this tool when users ask about:
- Current weather conditions
- Temperature
- Weather forecasts
Returns: Human-readable weather description
"""
api_key = os.getenv("OPENWEATHER_API_KEY")
url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}"
response = requests.get(url, timeout=5)
if response.status_code == 404:
return f"City '{city}' not found. Please check spelling."
if response.status_code != 200:
raise Exception(f"API returned status {response.status_code}")
data = response.json()
temp_f = (data["main"]["temp"] - 273.15) * 9/5 + 32
condition = data["weather"][0]["description"]
return f"Weather in {city}: {condition.capitalize()}, {temp_f:.1f}°F"
weather_tool = create_tool(
name="get_weather",
description="""Fetches current weather data from OpenWeatherMap API.
Use this tool when users ask about:
- Current weather conditions
- Temperature
- Weather forecasts
Returns: Human-readable weather description""",
input_parameters={"city": str},
call_function=fetch_weather,
on_error="return_error" # Return error message, don't crash
)
```
### Example 3: File Processing
```python
from peargent import create_tool
import os
def analyze_file(filepath: str) -> dict:
"""
Analyzes a text file and returns metadata and preview.
Use this tool when users want to:
- Check file size
- See file contents preview
- Count lines in a file
Returns: Dict with file metadata
"""
if not os.path.exists(filepath):
raise FileNotFoundError(f"File '{filepath}' does not exist")
size = os.path.getsize(filepath)
with open(filepath, 'r', encoding='utf-8') as f:
lines = f.readlines()
preview = ''.join(lines[:5])
return {
"filename": os.path.basename(filepath),
"size_bytes": size,
"line_count": len(lines),
"preview": preview
}
file_tool = create_tool(
name="analyze_file",
description="""Analyzes a text file and returns metadata and preview.
Use this tool when users want to:
- Check file size
- See file contents preview
- Count lines in a file
Returns: Dict with file metadata""",
input_parameters={"filepath": str},
call_function=analyze_file,
on_error="return_error"
)
# Usage
result = file_tool.run({"filepath": "/path/to/file.txt"})
print(f"File: {result['filename']}")
print(f"Size: {result['size_bytes']} bytes")
print(f"Lines: {result['line_count']}")
```
## Tool Building Checklist
Use this checklist when creating production tools:
### Design Phase
* [ ] Tool has a single, clear responsibility
* [ ] Input parameters are minimal and well-typed
* [ ] Docstring explains **when** to use the tool, not just what it does
* [ ] Examples included in docstring for clarity
### Implementation Phase
* [ ] All parameters have type hints
* [ ] Function handles errors gracefully (try-except)
* [ ] Tool is idempotent (same input = same output)
* [ ] No hidden global state
### Configuration Phase
* [ ] `on_error` strategy chosen based on criticality
### Testing Phase
* [ ] Tool works with valid inputs
* [ ] Tool handles invalid inputs gracefully
## Common Patterns
### Pattern 1: Multi-Step Tool
```python
from pydantic import BaseModel, Field
from peargent import create_tool
class OrderResult(BaseModel):
order_id: str
status: str
total: float
items: list[str]
def process_order(customer_id: str, items: list[str]) -> OrderResult:
"""
Processes a customer order through multiple steps.
Steps:
1. Validate customer exists
2. Check inventory for all items
3. Calculate total with taxes
4. Create order record
5. Update inventory
Returns: OrderResult with order details
"""
# Step 1: Validate customer
customer = db.get_customer(customer_id)
if not customer:
raise ValueError(f"Customer {customer_id} not found")
# Step 2: Check inventory
for item in items:
if not inventory.check_availability(item):
raise ValueError(f"Item {item} out of stock")
# Step 3: Calculate total
total = sum(catalog.get_price(item) for item in items)
tax = total * 0.0725
final_total = total + tax
# Step 4: Create order
order_id = db.create_order(customer_id, items, final_total)
# Step 5: Update inventory
for item in items:
inventory.decrement(item)
return OrderResult(
order_id=order_id,
status="completed",
total=final_total,
items=items
)
order_tool = create_tool(
name="process_order",
description="""Processes a customer order through multiple steps.
Steps:
1. Validate customer exists
2. Check inventory for all items
3. Calculate total with taxes
4. Create order record
5. Update inventory
Returns: OrderResult with order details""",
input_parameters={
"customer_id": str,
"items": list
},
call_function=process_order,
on_error="raise" # Critical operation - must succeed
)
```
### Pattern 2: Tool Composition
Break complex operations into smaller tools:
```python
# Small, focused tools
search_tool = create_tool(
name="search_products",
description="Searches products by keyword",
input_parameters={"query": str},
call_function=search_products
)
filter_tool = create_tool(
name="filter_by_price",
description="Filters products by price range",
input_parameters={"products": list, "min_price": float, "max_price": float},
call_function=filter_by_price
)
sort_tool = create_tool(
name="sort_products",
description="Sorts products by field",
input_parameters={"products": list, "sort_by": str},
call_function=sort_products
)
# Agent uses tools in sequence
agent = create_agent(
name="ProductAgent",
persona="You help users find products. Use multiple tools in sequence.",
tools=[search_tool, filter_tool, sort_tool]
)
# User: "Show me laptops under $1000, sorted by price"
# Agent will:
# 1. Call search_tool(query="laptops")
# 2. Call filter_tool(products=results, max_price=1000)
# 3. Call sort_tool(products=filtered, sort_by="price")
```
## When to Create Custom Tools
### ✅ Good Use Cases:
* **Domain-Specific Logic**: Calculating pricing based on your company's unique rules
* **Internal APIs**: Fetching data from your private microservices
* **Database Operations**: Querying your application's database
* **Complex Workflows**: Triggering multi-step processes (e.g., "onboard\_new\_employee")
* **External Integrations**: Calling third-party APIs (Stripe, Twilio, etc.)
* **File Operations**: Reading, writing, or processing files
* **Business Logic**: Tax calculations, shipping estimates, etc.
### ❌ Poor Use Cases:
* **Generic Operations**: Use existing tools (web search, calculations)
* **One-Off Tasks**: Write regular Python code instead
* **State Management**: Don't use tools to track conversation state (use history instead)
* **Pure Computation**: Simple math doesn't need a tool (LLM can do it)
## Summary
**Building Great Tools:**
1. **Clear Docstrings** - Explain *when* to use the tool, not just *what* it does
2. **Type Everything** - Full type hints for parameters and returns
3. **Handle Errors** - Choose appropriate `on_error` strategy
4. **Keep It Simple** - One responsibility per tool, idempotent when possible
**Essential Parameters:**
* `name`, `description`, `input_parameters`, `call_function` - Always required
* `on_error` - `"raise"`, `"return_error"`, or `"return_none"`
Start with simple tools and add advanced features as needed!
# Writing Effective Personas
Writing Effective Personas
Learn how to craft powerful system prompts that define your agent's behavior
The `persona` parameter in Peargent is your agent's system prompt—the instructions that define how your agent behaves, speaks, and solves problems. A well-crafted persona can mean the difference between a generic bot and a highly effective specialist.
**Your persona is sent with EVERY request**, so make it count!
## The Anatomy of a Strong Persona
A good persona should cover three key areas: **Identity**, **Capabilities**, and **Constraints**.
### 1. Identity (Who are you?)
Define the agent's role, expertise, and communication style.
#### Weak Identity
```python
agent = create_agent(
name="Assistant",
persona="You are a helpful assistant.", # ❌ Too generic!
model=groq("llama-3.3-70b-versatile")
)
```
#### Strong Identity
```python
agent = create_agent(
name="DevOpsExpert",
persona="""You are a Senior DevOps Engineer with 10 years of experience in
Kubernetes, AWS, and CI/CD pipelines.
Communication Style:
- Speak concisely and technically
- Use industry-standard terminology
- Always prioritize security and reliability
- Provide specific commands and configurations when relevant""", # ✅ Clear role & expertise!
model=groq("llama-3.3-70b-versatile")
)
```
**Key Elements:**
* **Expertise**: Define the agent's domain knowledge
* **Experience Level**: Senior, junior, specialist, generalist
* **Communication Style**: Concise, verbose, technical, friendly
* **Priorities**: What matters most (security, speed, cost, UX)
### 2. Capabilities (What can you do?)
Explicitly state what the agent is good at and how it should approach tasks.
```python
persona = """You are a Python Code Reviewer specializing in performance optimization.
Your Capabilities:
- Analyze code for time and space complexity
- Identify bottlenecks and inefficiencies
- Suggest algorithmic improvements
- Recommend appropriate data structures
- Profile memory usage patterns
Your Approach:
1. Read the code thoroughly
2. Identify the most critical performance issues first
3. Provide specific, actionable recommendations
4. Include code examples for complex changes
5. Explain the performance impact of each suggestion"""
```
**Benefits:**
* Agent knows exactly what it's supposed to do
* Consistent behavior across requests
* Clear scope of responsibilities
### 3. Constraints (What should you NOT do?)
Set boundaries to prevent hallucinations, unwanted behaviors, or scope creep.
```python
persona = """You are a Database Administrator assistant.
Constraints:
- Do NOT execute any DELETE or DROP commands without explicit user confirmation
- Do NOT modify production databases directly—always suggest backup strategies first
- Do NOT recommend solutions using databases you're unfamiliar with
- If unsure about a command's impact, ask for clarification instead of guessing
- Never assume data can be recovered—always confirm backup procedures exist"""
```
**Common Constraints:**
* Safety: "Never execute destructive commands without confirmation"
* Scope: "Do not provide legal or medical advice"
* Accuracy: "If unsure, say 'I don't know' instead of guessing"
* Tool Usage: "Always use tools instead of making up data"
## Real-World Persona Examples
### Example 1: Code Reviewer Agent
```python
from peargent import create_agent
from peargent.models import groq
code_reviewer = create_agent(
name="CodeReviewBot",
description="Expert Python code reviewer",
persona="""You are an expert Python Code Reviewer with expertise in:
- Software architecture and design patterns
- Performance optimization
- Security vulnerabilities (SQL injection, XSS, etc.)
- PEP 8 style compliance
- Testing best practices
Your Review Process:
1. **Security First**: Flag any potential security risks immediately
2. **Correctness**: Identify logic errors and edge cases
3. **Performance**: Suggest optimizations for slow code
4. **Readability**: Recommend style improvements
5. **Testing**: Highlight untested code paths
Your Communication:
- Be constructive, not critical
- Explain *why* a change is needed, not just *what* to change
- Provide code snippets showing the fix
- Use examples to illustrate concepts
- Prioritize issues: Critical > Major > Minor
Constraints:
- Do not rewrite entire files unless absolutely necessary
- Focus on the specific function or block in question
- Do not suggest libraries that don't exist
- If code is correct, say so—don't invent issues""",
model=groq("llama-3.3-70b-versatile")
)
# Usage
review = code_reviewer.run("""
def calculate_total(prices):
total = 0
for price in prices:
total = total + price
return total
""")
```
### Example 2: Customer Support Agent
```python
support_agent = create_agent(
name="SupportBot",
description="Friendly customer support specialist",
persona="""You are a friendly and empathetic Customer Support Specialist.
Your Mission:
Help customers resolve issues quickly while maintaining a positive experience.
Your Personality:
- Warm and approachable
- Patient with frustrated customers
- Proactive in offering solutions
- Clear and non-technical in explanations
Response Structure:
1. Acknowledge the customer's issue with empathy
2. Ask clarifying questions if needed
3. Provide step-by-step solution
4. Verify the solution worked
5. Offer additional help if needed
Constraints:
- Never promise refunds without checking policy
- Do not share other customers' information
- Escalate to human support for: billing disputes, legal threats, abuse
- Do not make up features that don't exist
- If you can't help, admit it and escalate
Example Tone:
"I understand how frustrating that must be! Let's get this sorted out for you.
Can you tell me what error message you're seeing?"
NOT: "Error detected. Follow these steps: ..."
""",
model=groq("llama-3.3-70b-versatile")
)
```
### Example 3: Data Analyst Agent
```python
analyst_agent = create_agent(
name="DataAnalyst",
description="Statistical data analysis expert",
persona="""You are an expert Data Analyst with strong statistical and analytical skills.
Your Expertise:
- Statistical analysis (mean, median, variance, correlation)
- Data visualization recommendations
- Trend identification and forecasting
- Hypothesis testing
- Data quality assessment
Your Analysis Process:
1. Understand the business question
2. Examine data structure and quality
3. Perform relevant statistical calculations
4. Identify patterns, trends, and anomalies
5. Provide actionable insights with confidence levels
Your Deliverables:
- Executive summary (1-2 sentences)
- Key findings (bullet points)
- Statistical evidence (numbers, percentages)
- Visualizations recommendations
- Next steps or recommendations
Communication Style:
- Explain statistics in business terms
- Always include context with numbers
- Highlight uncertainty and confidence levels
- Use analogies for complex concepts
Constraints:
- Do not claim causation without proper analysis
- Always mention sample size and data quality
- Flag potential biases in the data
- If data is insufficient, say so clearly""",
model=groq("llama-3.3-70b-versatile")
)
```
### Example 4: Creative Writer Agent
```python
writer_agent = create_agent(
name="CreativeWriter",
description="Imaginative storyteller",
persona="""You are a creative writer who weaves compelling and imaginative stories.
Your Writing Style:
- Vivid, descriptive language
- Engaging narrative hooks
- Strong character development
- Unexpected plot twists
- Emotional depth
Story Structure:
1. Hook: Grab attention immediately
2. Setting: Paint the scene
3. Conflict: Introduce tension
4. Development: Build the narrative
5. Resolution: Satisfying conclusion
Your Approach:
- Show, don't tell
- Use sensory details (sight, sound, smell, touch, taste)
- Vary sentence length for rhythm
- Create memorable characters with distinct voices
- End with impact
Constraints:
- Keep stories appropriate for general audiences unless specified
- Respect character consistency
- Do not break the fourth wall unless intentional
- Avoid clichés and overused tropes
Respond directly with your story—do not use tools or JSON formatting.""",
model=groq("llama-3.3-70b-versatile")
)
```
### Example 5: Research Specialist
```python
researcher_agent = create_agent(
name="Researcher",
description="Meticulous data researcher",
persona="""You are a meticulous data researcher who specializes in gathering
comprehensive information.
Your Mission:
Collect relevant, accurate, and well-organized data from available sources.
Research Methodology:
1. Understand the research question
2. Use all available tools to gather data
3. Verify information from multiple sources when possible
4. Organize findings logically
5. Cite sources and provide context
Data Collection:
- Use tools systematically (don't skip available resources)
- Extract key facts, numbers, and quotes
- Note data quality and reliability
- Highlight conflicting information
- Preserve source attribution
Presentation Style:
- Structured and organized
- Factual and objective
- Comprehensive but not verbose
- Clear headings and bullet points
- Source citations
Constraints:
- Focus purely on data collection—do not analyze or interpret
- Do not editorialize or inject opinions
- If data is unavailable, state this clearly
- Do not fabricate or guess data points
- Analysis is for other specialists, not you""",
model=groq("llama-3.3-70b-versatile")
)
```
## Persona Optimization Strategies
### Strategy 1: Be Specific About Tool Usage
If your agent has tools, tell it exactly when and how to use them.
```python
persona = """You are a helpful assistant with access to a product database.
Tool Usage Guidelines:
- ALWAYS use the search_database tool when users ask about products
- Do NOT make up product information
- If search returns no results, tell the user—don't invent products
- Use tools multiple times if needed to answer complex questions
- Combine tool results in your response
Example:
User: "Do you have red shirts?"
You: Call search_database(query="red shirts") then report findings"""
```
### Strategy 2: Use Markdown for Structure
LLMs understand structured text well. Use formatting in your persona string.
```python
persona = """You are a Technical Writer.
## Your Mission
Create clear, concise documentation for software developers.
## Writing Principles
- **Clarity**: Use simple language
- **Completeness**: Cover all edge cases
- **Consistency**: Follow established patterns
- **Code Examples**: Always include working code
## Document Structure
1. Overview (what and why)
2. Prerequisites
3. Step-by-step instructions
4. Code examples
5. Troubleshooting
## Tone
Professional but approachable, like a helpful colleague."""
```
### Strategy 3: Provide Examples in Context
Show the agent what good responses look like.
```python
persona = """You are a Python tutor who explains concepts clearly.
Example Interaction:
Student: "What is a list comprehension?"
Good Response:
"A list comprehension is a concise way to create lists in Python.
Basic syntax: [expression for item in iterable]
Example:
# Traditional loop
squares = []
for x in range(5):
squares.append(x**2)
# List comprehension (same result)
squares = [x**2 for x in range(5)]
Result: [0, 1, 4, 9, 16]
Use list comprehensions when you want to transform each item in a list."
Bad Response: "List comprehensions create lists efficiently."
Always explain like the good example: concept + syntax + code + result."""
```
### Strategy 4: Iterate and Refine
Don't expect perfection on the first try. Test your persona and refine it.
```python
# Version 1: Too vague
persona_v1 = "You are a helpful coding assistant."
# Version 2: More specific
persona_v2 = "You are a Python expert who helps with code debugging."
# Version 3: Comprehensive (final)
persona_v3 = """You are a Python debugging expert.
When users share code with errors:
1. Identify the error type and line number
2. Explain what's causing the error
3. Show the corrected code
4. Explain why the fix works
Always:
- Test your suggested fixes mentally before sharing
- Explain in simple terms
- Provide complete, runnable code
- Highlight the changes you made"""
```
## Common Persona Pitfalls
### Pitfall 1: Too Generic
```python
# Bad: Agent doesn't know its purpose
persona = "You are a helpful AI assistant."
# Good: Clear purpose and expertise
persona = "You are a SQL database expert who helps optimize queries for PostgreSQL."
```
### Pitfall 2: Too Verbose
Remember: Your persona is sent with **every single request**!
```python
# Bad: 300+ tokens wasted per request
persona = """You are a highly knowledgeable and extremely helpful AI assistant
who always strives to provide the most comprehensive and detailed answers possible.
You should always be polite, courteous, and respectful in your interactions.
You have expertise in many domains including science, technology, arts, history,
mathematics, literature, philosophy, psychology, sociology, and more.
You should always think carefully before responding and provide well-structured
answers. You should use examples when appropriate and explain concepts clearly
and thoroughly. You should be patient with users and never make them feel bad
for not understanding something...""" # ❌ ~200 tokens!
# Good: Concise and focused
persona = """You are a helpful assistant with expertise in science and technology.
Be clear, concise, and accurate. Provide examples when helpful.""" # ✅ ~20 tokens!
```
**Cost Impact:** 180 tokens saved × 1000 requests = 180,000 tokens saved!
### Pitfall 3: Conflicting Instructions
```python
# Bad: Contradictory instructions
persona = """You are a concise assistant.
Always provide detailed, comprehensive explanations with examples.
Keep responses brief.""" # ❌ Confusing!
# Good: Clear, consistent instructions
persona = """You are a concise technical assistant.
Provide brief but complete answers with one example when needed.""" # ✅ Clear!
```
### Pitfall 4: No Constraints
```python
# Bad: No safety boundaries
persona = "You are a database admin who helps with SQL queries."
# Agent might run DROP TABLE without confirmation!
# Good: Safety constraints
persona = """You are a database admin assistant.
Safety Rules:
- NEVER execute DELETE, DROP, or TRUNCATE without explicit confirmation
- Always suggest BACKUP before destructive operations
- Test queries on small datasets first
- Explain potential impact of each command"""
```
## Multi-Agent Persona Design
When building agent pools, each agent should have a distinct, focused persona.
```python
from peargent import create_agent, create_pool
from peargent.models import groq
# Agent 1: Researcher (data collection)
researcher = create_agent(
name="Researcher",
description="Data collection specialist",
persona="""You are a meticulous data researcher.
Your ONLY job: Gather relevant data using available tools.
- Use tools systematically
- Collect comprehensive information
- Organize findings clearly
- Do NOT analyze or interpret data
- Do NOT provide recommendations
Present your findings and stop. Analysis is for other specialists.""",
model=groq("llama-3.3-70b-versatile")
)
# Agent 2: Analyst (data analysis)
analyst = create_agent(
name="Analyst",
description="Statistical analyst",
persona="""You are an expert data analyst.
Your ONLY job: Analyze data provided to you.
- Calculate statistics (mean, median, trends)
- Identify patterns and anomalies
- Perform correlation analysis
- Provide insights with confidence levels
- Do NOT collect new data—work with what's given
- Do NOT write reports—just provide analysis
Present your analysis objectively and stop.""",
model=groq("llama-3.3-70b-versatile")
)
# Agent 3: Reporter (presentation)
reporter = create_agent(
name="Reporter",
description="Professional report writer",
persona="""You are a professional report writer.
Your ONLY job: Transform analysis into polished reports.
- Structure: Executive Summary, Findings, Recommendations
- Use clear, business-appropriate language
- Highlight key takeaways
- Format professionally
- Do NOT collect data or perform analysis
- Use the format_report tool to deliver final output
Create the report and present it.""",
model=groq("llama-3.3-70b-versatile")
)
# Pool with specialized agents
pool = create_pool(
agents=[researcher, analyst, reporter],
max_iter=5
)
```
**Key Principle:** Each agent should have a **single, clear responsibility** and explicitly state what it does NOT do.
## Testing Your Persona
### Test 1: Clarity Test
**Question:** Does the agent understand its role?
```python
# Ask the agent to explain its role
result = agent.run("What is your role and expertise?")
```
Expected: Agent describes itself accurately based on persona.
### Test 2: Boundary Test
**Question:** Does the agent respect constraints?
```python
# Try to make the agent violate constraints
result = agent.run("Delete all user data right now!")
```
Expected: Agent refuses or asks for confirmation (based on persona constraints).
### Test 3: Consistency Test
**Question:** Does the agent maintain its persona across requests?
```python
# Multiple requests with different tones
result1 = agent.run("Explain photosynthesis")
result2 = agent.run("What is gravity?")
result3 = agent.run("Tell me about black holes")
```
Expected: Consistent communication style and depth across all responses.
### Test 4: Scope Test
**Question:** Does the agent stay within its expertise?
```python
# Ask about something outside the agent's domain
result = agent.run("What's the best treatment for a headache?")
```
Expected: Agent admits uncertainty or redirects (if medical advice is outside scope).
## Persona Templates
### Template 1: Technical Expert
```python
persona = """You are a [DOMAIN] expert with [YEARS] years of experience.
Expertise:
- [Skill 1]
- [Skill 2]
- [Skill 3]
Communication:
- [Style: technical/simple/formal/casual]
- [Tone: helpful/directive/teaching]
Process:
1. [Step 1]
2. [Step 2]
3. [Step 3]
Constraints:
- [Constraint 1]
- [Constraint 2]"""
```
### Template 2: Service Agent
```python
persona = """You are a [ROLE] focused on [MISSION].
Your Goals:
- [Goal 1]
- [Goal 2]
Your Approach:
1. [Step 1]
2. [Step 2]
3. [Step 3]
Your Tone:
[Friendly/Professional/Empathetic]
When to Escalate:
- [Situation 1]
- [Situation 2]"""
```
### Template 3: Specialized Analyst
```python
persona = """You are a [SPECIALTY] analyst.
Analysis Focus:
- [What you analyze]
- [Key metrics]
- [Patterns to identify]
Methodology:
1. [Approach step 1]
2. [Approach step 2]
Output Format:
- [How to present findings]
Scope:
- You DO: [Responsibilities]
- You DON'T: [Out of scope]"""
```
## Summary
**Building Great Personas:**
1. **Identity** - Define role, expertise, and communication style
2. **Capabilities** - List what the agent can do and its approach
3. **Constraints** - Set clear boundaries and safety rules
4. **Conciseness** - Keep it brief (personas are sent with every request)
5. **Structure** - Use markdown, bullet points, and headers
6. **Examples** - Show what good responses look like
7. **Specificity** - Be precise about tool usage and behavior
**Persona Checklist:**
* [ ] Role and expertise clearly defined
* [ ] Communication style specified
* [ ] Capabilities explicitly listed
* [ ] Constraints and boundaries set
* [ ] Tool usage guidelines included (if applicable)
* [ ] Examples provided for complex behaviors
* [ ] Concise (\< 150 tokens for cost optimization)
* [ ] Tested with various inputs
**Remember:** Your persona is the foundation of agent behavior. Invest time in crafting it well, and your agent will perform consistently and reliably!
# History Best Practices
## Use Metadata for Search and Filtering
```python
# Tag threads with searchable metadata
history.create_thread(metadata={
"user_id": "alice",
"topic": "technical_support",
"priority": "high",
"created_by": "web_app",
"tags": ["billing", "bug_report"]
})
# Later: Find threads by metadata
all_threads = history.list_threads()
high_priority = []
for thread_id in all_threads:
thread = history.get_thread(thread_id)
if thread.metadata.get("priority") == "high":
high_priority.append(thread_id)
```
## Use Message Metadata for Tracking
```python
history.add_user_message(
"Process this document",
metadata={
"source": "api",
"user_ip": "192.168.1.1",
"request_id": "req_123",
"timestamp_ms": 1234567890
}
)
history.add_assistant_message(
"Document processed successfully",
agent="DocumentProcessor",
metadata={
"model": "gpt-4o",
"tokens_used": 1250,
"latency_ms": 2340,
"cost_usd": 0.025
}
)
```
## Implement Cleanup Policies
```python
from datetime import datetime, timedelta
def cleanup_old_threads(history, days=30):
"""Delete threads older than specified days."""
cutoff = datetime.now() - timedelta(days=days)
all_threads = history.list_threads()
deleted = 0
for thread_id in all_threads:
thread = history.get_thread(thread_id)
if thread.created_at < cutoff:
history.delete_thread(thread_id)
deleted += 1
return deleted
# Run cleanup
deleted_count = cleanup_old_threads(history, days=90)
print(f"Deleted {deleted_count} old threads")
```
## Export Conversations
```python
import json
def export_thread(history, thread_id, filename):
"""Export thread to JSON file."""
thread = history.get_thread(thread_id)
if not thread:
return False
with open(filename, 'w') as f:
json.dump(thread.to_dict(), f, indent=2)
return True
# Export specific conversation
export_thread(history, thread_id, "conversation_export.json")
# Import conversation
def import_thread(history, filename):
"""Import thread from JSON file."""
from peargent.storage import Thread
with open(filename, 'r') as f:
data = json.load(f)
thread = Thread.from_dict(data)
# Create thread in history
new_thread_id = history.create_thread(metadata=thread.metadata)
# Add all messages
for msg in thread.messages:
if msg.role == "user":
history.add_user_message(msg.content, metadata=msg.metadata)
elif msg.role == "assistant":
history.add_assistant_message(msg.content, agent=msg.agent, metadata=msg.metadata)
elif msg.role == "tool":
history.add_tool_message(msg.tool_call, agent=msg.agent, metadata=msg.metadata)
return new_thread_id
```
## Context Window Monitoring
```python
def should_manage_context(history, threshold=20):
"""Check if context management is needed."""
count = history.get_message_count()
if count > threshold:
print(f"⚠️ Context window full: {count}/{threshold} messages")
return True
else:
print(f"✓ Context OK: {count}/{threshold} messages")
return False
# Monitor before agent runs
if should_manage_context(history, threshold=25):
history.manage_context_window(
model=groq("llama-3.1-8b-instant"),
max_messages=25,
strategy="smart"
)
```
# Practical Playbooks
Practical Playbooks
Actionable guides and strategies for building real-world agents
Welcome to the Practical Playbooks. This section contains hands-on guides to help you master specific aspects of Peargent development.
## Available Playbooks
### **[Writing Effective Personas](/docs/practical-playbooks/effective-personas)**
Learn how to craft powerful system prompts that define your agent's behavior, tone, and constraints.
### **[Creating Custom Tooling](/docs/practical-playbooks/custom-tooling)**
Best practices for building robust, reusable tools that LLMs can use reliably.
### **[Optimizing for Cost](/docs/practical-playbooks/cost-optimization)**
Strategies to control token usage and reduce API costs without sacrificing performance.
# Agent Output
Structured Output for Agent
Build agents that return reliable, schema-validated structured responses.
Structured output means the **[Agent](/docs/agents)** returns its answer in a format you define, such as a dictionary, JSON-like object, or typed schema.
Instead of replying with free-form text, the agent fills the exact fields and structure you specify.
## Why Structured Output?
Structured output is useful because:
* It gives consistent and predictable responses
* Your code can easily read and use the output without parsing text
* It prevents the model from adding extra text or changing the format
* It makes agents reliable for automation, APIs, databases, UI generation, and workflows
## Enforcing a simple schema
We will be using pydantic package to create the output schema. Pydantic is a data validation and settings management using Python type annotations.
More about pydantic: [https://docs.pydantic.dev/latest/](https://docs.pydantic.dev/latest/)
Make sure to install pydantic package `pip install pydantic`.
```python
from pydantic import BaseModel, Field
from peargent import create_agent
from peargent.models import openai
# 1. Define your schema// [!code highlight:4]
class Summary(BaseModel):
title: str = Field(description="Short title for the summary")
points: list[str] = Field(description="Key points extracted from the text")
# 2. Create agent with structured output
agent = create_agent(
name="Summarizer",
description="Summarizes long text into structured key points.",
persona="You are a precise summarizer.",
model=openai("gpt-4o"),
output_schema=Summary, # ← enforce structured output // [!code highlight]
)
# 3. Run the agent
result = agent.run("Long text to summarize")
print(result)
```
Output (always structured):
```json
{
"title": "Understanding Black Holes",
"points": [
"They form when massive stars collapse.",
"Their gravity is extremely strong.",
"Nothing can escape once inside the event horizon."
]
}
```
## Schema with Custom Validators
Sometimes models return values that don't match the schema. Peargent integrates with **Pydantic validators** to enforce rules like rejecting incorrect values or cleaning fields. If validation fails, Peargent automatically retries until the output is valid (respecting `max_retries`).
This is particularly useful for ensuring that generated data meets strict business logic requirements, such as validating email formats, checking price ranges, or ensuring strings meet specific length constraints.
```python
from pydantic import BaseModel, Field, field_validator
from peargent import create_agent
from peargent.models import openai
# Define a simple structured output with validators
class Product(BaseModel):
name: str = Field(description="Product name")
price: float = Field(description="Price in USD", ge=0)
category: str = Field(description="Product category")
# Validator: ensure product name is not too generic // [!code highlight:22]
@field_validator("name")
@classmethod
def name_not_generic(cls, v):
"""
Ensure the product name is not overly generic (e.g., 'item', 'product', 'thing').
Helps maintain meaningful and descriptive product naming.
"""
forbidden = ["item", "product", "thing"]
if v.lower() in forbidden:
raise ValueError("Product name is too generic")
return v
# Validator: enforce category capitalization
@field_validator("category")
@classmethod
def category_must_be_titlecase(cls, v):
"""
Automatically convert the product category into Title Case
to maintain consistent formatting across all entries.
"""
return v.title()
# Create agent with structured output validation
agent = create_agent(
name="ProductGenerator",
description="Generates product details",
persona="You describe products clearly and accurately.",
model=openai("gpt-4o"),
output_schema=Product,
max_retries=3
)
product = agent.run("Generate a new gadget idea for travelers")
print(product)
```
**Why validator docstrings matter:**
Docstrings are included in the prompt sent to the model. They explain your custom validation rules in natural language, helping the LLM avoid mistakes before they happen. This drastically reduces failed validations, retries, and extra API calls, saving cost and improving reliability.
## Nested Output Schema
You can also nest multiple Pydantic models inside each other.
This allows your agent to return clean, hierarchical, and well-organized structured output, perfect for complex data like profiles, products, events, or summaries.
```python
from pydantic import BaseModel, Field
from typing import List
from peargent import create_agent
from peargent.models import openai
# ----- Nested Models ----- // [!code highlight:5]
class Address(BaseModel):
street: str = Field(description="Street name")
city: str = Field(description="City name")
country: str = Field(description="Country name")
class UserProfile(BaseModel):
name: str = Field(description="Full name of the user")
age: int = Field(description="Age of the user", ge=0, le=120)
email: str = Field(description="Email address")
# Nesting Address schema // [!code highlight:2]
address: Address = Field(description="Residential address")
hobbies: List[str] = Field(description="List of hobbies")
# ----- Create Agent with Nested Schema -----
agent = create_agent(
name="ProfileBuilder",
description="Builds structured user profiles",
persona="You extract and organize user information accurately.",
model=openai("gpt-4o"),
output_schema=UserProfile
)
profile = agent.run("Generate a profile for John Doe who lives in London.")
print(profile)
```
**Output shape:**
```json
{
"name": "John Doe",
"age": 32,
"email": "john.doe@example.com",
"address": {
"street": "221B Baker Street",
"city": "London",
"country": "United Kingdom"
},
"hobbies": ["reading", "cycling"]
}
```
## How Structured Output Works
### Output Schema Is Extracted
**Agent** first reads your **output\_schema** (Pydantic model) and extracts field names, types, required fields, and constraints (e.g., min\_length, ge, le). This forms the core **JSON schema** that the model must follow.
### Validator Docstrings Are Collected
Next, **Agent** scans your **Pydantic validators** and collects the **docstrings** you wrote inside them. These docstrings describe custom rules in natural language, such as “Name must not be generic” or “Price must be realistic”.
These **docstrings** are critical because:
* The LLM understands natural language rules
* It reduces retries (→ lower cost)
* It helps the model produce valid JSON on the first attempt
### Schema + Validator Rules Are Combined and Sent to the Model
**Agent** merges the JSON schema, field constraints, and validator docstrings into a single structured prompt.
At this point, the **Model** receives:
* The complete structure it must output
* Every validation rule it must follow
* Clear natural-language constraints
This ensures the **Model** is fully aware of what the final response should look like.
### Model Generates a Response
The Model now returns a JSON object that attempts to satisfy the full schema and all validation rules.
### Pydantic Validates the Response
Agent parses the JSON into your Pydantic model (`output_schema`), performing type checks, verifying missing fields, and running validator functions. If validation fails, **Agent** asks the **Model** to correct the response.
### Retry Loop (Until Valid Output)
If a validator rejects the response:
* Agent sends the error back to the **Model**
* The **Model** tries again
* Loop continues until max retries are reached
### Final Clean Pydantic Object Returned
After validation succeeds, Agent returns a fully typed, fully validated, and safe-to-use object.
# Tools Output
Structured Output for Tools
Build tools that return reliable, schema-validated structured outputs.
Structured output means the **[Tool](/docs/tools)** returns its result in a format you define, such as a dictionary, JSON-like object, or typed schema.
Instead of returning raw data, the tool validates and returns typed Pydantic model instances with guaranteed structure and correctness.
## Why Structured Output for Tools?
Structured output is useful because:
* It ensures tools return consistent, validated data
* Your agents can reliably use tool outputs without parsing errors
* It catches malformed API responses, database records, or external data early
* It provides type safety and IDE autocomplete for tool results
* It makes tools reliable for production systems, APIs, and complex workflows
## Validating Tool Output with Schema
We will be using pydantic package to validate tool outputs. Pydantic is a data validation and settings management using Python type annotations.
More about pydantic: [https://docs.pydantic.dev/latest/](https://docs.pydantic.dev/latest/)
Make sure to install pydantic package `pip install pydantic`.
```python
from pydantic import BaseModel, Field
from peargent import create_tool
# 1. Define your output schema// [!code highlight:4]
class WeatherData(BaseModel):
temperature: float = Field(description="Temperature in Fahrenheit")
condition: str = Field(description="Weather condition (e.g., Sunny, Cloudy)")
humidity: int = Field(description="Humidity percentage", ge=0, le=100)
# 2. Create tool with output validation
def get_weather(city: str) -> dict:
# Simulated API call
return {
"temperature": 72.5,
"condition": "Sunny",
"humidity": 45
}
weather_tool = create_tool(
name="get_weather",
description="Get current weather for a city",
input_parameters={"city": str},
call_function=get_weather,
output_schema=WeatherData, # ← validate tool output // [!code highlight]
)
# 3. Run the tool
result = weather_tool.run({"city": "San Francisco"})
print(result)
```
Output (validated Pydantic model):
```python
WeatherData(temperature=72.5, condition='Sunny', humidity=45)
# Access with type safety
print(result.temperature) # 72.5
print(result.condition) # "Sunny"
print(result.humidity) # 45
```
## Schema with Constraints and Validation
Tools can enforce strict validation rules using Pydantic field constraints. If the tool's raw output violates these constraints, validation will fail and the error will be handled based on the `on_error` parameter.
This is particularly useful for validating external API responses, database queries, or any tool that returns data from untrusted sources.
```python
from pydantic import BaseModel, Field, field_validator
from peargent import create_tool
# Define schema with constraints
class UserProfile(BaseModel):
user_id: int = Field(description="Unique user ID", gt=0)
username: str = Field(description="Username", min_length=3, max_length=20)
email: str = Field(description="Email address")
age: int = Field(description="User age", ge=0, le=150)
premium: bool = Field(description="Premium subscription status")
# Custom validator: email must contain @ symbol // [!code highlight:10]
@field_validator("email")
@classmethod
def validate_email(cls, v):
"""
Ensure email contains @ symbol.
This catches malformed email addresses from database or API.
"""
if "@" not in v:
raise ValueError("Invalid email format")
return v
# Tool that fetches user data
def fetch_user(user_id: int) -> dict:
# Simulated database query
return {
"user_id": user_id,
"username": "john_doe",
"email": "john@example.com",
"age": 28,
"premium": True
}
user_tool = create_tool(
name="fetch_user",
description="Fetch user profile from database",
input_parameters={"user_id": int},
call_function=fetch_user,
output_schema=UserProfile,
on_error="return_error" # Gracefully handle validation failures
)
# Use the tool
result = user_tool.run({"user_id": 123})
print(result)
```
## Nested Output Schema
You can nest multiple Pydantic models inside each other for complex tool outputs.
This is perfect for validating API responses, database records with relationships, or any hierarchical data structure.
```python
from pydantic import BaseModel, Field
from typing import List
from peargent import create_tool
# ----- Nested Models ----- // [!code highlight:10]
class Address(BaseModel):
street: str = Field(description="Street address")
city: str = Field(description="City name")
state: str = Field(description="State code")
zip_code: str = Field(description="ZIP code")
class PhoneNumber(BaseModel):
type: str = Field(description="Phone type: mobile, home, or work")
number: str = Field(description="Phone number")
class ContactInfo(BaseModel):
name: str = Field(description="Full name")
email: str = Field(description="Email address")
# Nested schemas // [!code highlight:3]
address: Address = Field(description="Mailing address")
phone_numbers: List[PhoneNumber] = Field(description="Contact phone numbers")
notes: str = Field(description="Additional notes", default="")
# ----- Create Tool with Nested Schema -----
def fetch_contact(contact_id: int) -> dict:
# Simulated CRM API call
return {
"name": "Alice Johnson",
"email": "alice@example.com",
"address": {
"street": "123 Main St",
"city": "San Francisco",
"state": "CA",
"zip_code": "94102"
},
"phone_numbers": [
{"type": "mobile", "number": "415-555-1234"},
{"type": "work", "number": "415-555-5678"}
],
"notes": "Preferred contact method: email"
}
contact_tool = create_tool(
name="fetch_contact",
description="Fetch contact information from CRM",
input_parameters={"contact_id": int},
call_function=fetch_contact,
output_schema=ContactInfo
)
contact = contact_tool.run({"contact_id": 456})
print(contact)
```
**Output shape:**
```json
{
"name": "Alice Johnson",
"email": "alice@example.com",
"address": {
"street": "123 Main St",
"city": "San Francisco",
"state": "CA",
"zip_code": "94102"
},
"phone_numbers": [
{"type": "mobile", "number": "415-555-1234"},
{"type": "work", "number": "415-555-5678"}
],
"notes": "Preferred contact method: email"
}
```
## How Structured Output Works for Tools
### Tool Executes
**Tool** calls the `call_function()` which returns raw data (dict, object, etc.) from an API, database, or computation.
### Output Schema Is Checked
If an **output\_schema** is provided, **Tool** proceeds to validation. Otherwise, the raw output is returned as-is.
### Pydantic Validates the Output
**Tool** attempts to convert the raw output into the Pydantic model, performing:
* Type checking (str, int, float, bool, etc.)
* Required field verification
* Constraint validation (ge, le, min\_length, max\_length)
* Custom validator execution (@field\_validator)
If the output is already a Pydantic model instance of the correct type, it passes validation immediately.
### Validation Success or Failure
**If validation succeeds:**
* **Tool** returns the validated Pydantic model instance
* Type-safe, guaranteed structure
* Ready to use in agent workflows
**If validation fails:**
* Error is handled based on `on_error` parameter
* If `max_retries > 0`, tool automatically retries execution
* Validation runs again on each retry
* See **[Error Handling](/docs/error-handling-in-tools)** for details
### Final Validated Output Returned
After successful validation, **Tool** returns a fully typed, fully validated Pydantic object ready for use by agents or downstream code.
# Accessing Traces
Accessing Traces
Retrieve and analyze trace data from your agents and pools
After running agents with tracing enabled, you can access the trace data to analyze execution details, costs, performance, and errors.
## Getting the Tracer
Get the global tracer instance to access stored data.
```python
from peargent.observability import get_tracer
tracer = get_tracer()
```
## Listing Traces
Retrieve a list of traces with optional filtering.
```python
traces = tracer.list_traces(
agent_name: str = None, # Filter by agent name
session_id: str = None, # Filter by session ID
user_id: str = None, # Filter by user ID
limit: int = 100 # Max number of traces to return
)
```
## Getting a Single Trace
Retrieve a full trace object by its unique ID.
```python
trace = tracer.get_trace(trace_id: str)
```
## Trace Object Structure
The `Trace` object contains the following properties:
| Property | Type | Description |
| :------------- | :----------- | :--------------------------------------- |
| `id` | `str` | Unique identifier for the trace. |
| `agent_name` | `str` | Name of the agent that executed. |
| `session_id` | `str` | Session ID (if set). |
| `user_id` | `str` | User ID (if set). |
| `input_data` | `Any` | Input provided to the agent. |
| `output` | `Any` | Final output from the agent. |
| `start_time` | `datetime` | When execution started. |
| `end_time` | `datetime` | When execution ended. |
| `duration_ms` | `float` | Total duration in milliseconds. |
| `total_tokens` | `int` | Total tokens used (prompt + completion). |
| `total_cost` | `float` | Total cost in USD. |
| `error` | `str` | Error message if execution failed. |
| `spans` | `List[Span]` | List of operations within the trace. |
**Example:**
```python
print(f"Trace ID: {trace.id}")
```
## Span Object Structure
The `Span` object represents a single operation (LLM call, tool execution, etc.):
| Property | Type | Description |
| :------------ | :--------- | :--------------------------------------------- |
| `span_type` | `str` | Type of span: `"llm"`, `"tool"`, or `"agent"`. |
| `name` | `str` | Name of the model or tool. |
| `start_time` | `datetime` | Start timestamp. |
| `end_time` | `datetime` | End timestamp. |
| `duration_ms` | `float` | Duration in milliseconds. |
| `cost` | `float` | Cost of this specific operation. |
**Example:**
```python
print(f"Span duration: {span.duration_ms}ms")
```
**LLM Spans only:**
| Property | Type | Description |
| :------------------ | :---- | :---------------------------- |
| `llm_model` | `str` | Model name (e.g., "gpt-4o"). |
| `llm_prompt` | `str` | The prompt sent to the model. |
| `llm_response` | `str` | The response received. |
| `prompt_tokens` | `int` | Token count for prompt. |
| `completion_tokens` | `int` | Token count for completion. |
**Example:**
```python
print(f"Model used: {span.llm_model}")
```
**Tool Spans only:**
| Property | Type | Description |
| :------------ | :----- | :---------------------------- |
| `tool_name` | `str` | Name of the tool executed. |
| `tool_args` | `dict` | Arguments passed to the tool. |
| `tool_output` | `str` | Output returned by the tool. |
**Example:**
```python
print(f"Tool output: {span.tool_output}")
```
## Printing Traces
Print traces to the console for debugging.
```python
tracer.print_traces(
limit: int = 10, # Number of traces to print
format: str = "table" # "table", "json", "markdown", or "terminal"
)
```
## Printing Summary
Print a high-level summary of usage and costs.
```python
tracer.print_summary(
agent_name: str = None,
session_id: str = None,
user_id: str = None,
limit: int = None
)
```
## Aggregate Statistics
Get a dictionary of aggregated metrics programmatically.
```python
stats = tracer.get_aggregate_stats(
agent_name: str = None,
session_id: str = None,
user_id: str = None,
limit: int = None
)
```
**Returned Stats Dictionary:**
| Key | Type | Description |
| :--------------------- | :---------- | :--------------------------------------- |
| `total_traces` | `int` | Total number of traces matching filters. |
| `total_cost` | `float` | Total cost in USD. |
| `total_tokens` | `int` | Total tokens used. |
| `total_duration` | `float` | Total duration in ms. |
| `total_llm_calls` | `int` | Total number of LLM calls. |
| `total_tool_calls` | `int` | Total number of tool executions. |
| `avg_cost_per_trace` | `float` | Average cost per trace. |
| `avg_tokens_per_trace` | `float` | Average tokens per trace. |
| `avg_duration_ms` | `float` | Average duration per trace. |
| `agents_used` | `List[str]` | List of unique agent names found. |
**Example:**
```python
print(f"Total cost: ${stats['total_cost']}")
```
## What's Next?
**[Cost Tracking](/docs/tracing-and-observability/cost-tracking)**
Deep dive into cost analysis, token counting, and optimization strategies.
# Controlling Tracing
Controlling Tracing
Enable or disable tracing at the global and agent level
Peargent provides flexible control over tracing behavior. You can enable tracing globally with `enable_tracing()` and selectively opt agents in or out using the `tracing` parameter.
## How Tracing Works
The interaction between `enable_tracing()` and the `tracing` parameter determines whether an agent is traced:
| Global (`enable_tracing()`) | Agent/Pool (`tracing=`) | Result | Explanation |
| --------------------------- | ----------------------- | ------------ | --------------------------------------------------- |
| ❌ Not called | Not specified | ❌ No tracing | Default: tracing disabled |
| ❌ Not called | `tracing=True` | ❌ No tracing | Agent wants tracing but no global tracer configured |
| ✅ Called | Not specified | ✅ Traced | Agent inherits global tracing |
| ✅ Called | `tracing=True` | ✅ Traced | Agent explicitly opts in |
| ✅ Called | `tracing=False` | ❌ No tracing | Agent explicitly opts out |
| ✅ Called (`enabled=False`) | `tracing=True` | ❌ No tracing | Global `enabled=False` takes precedence |
## Global Control
Use `enable_tracing()` to control the master switch.
```python
from peargent.observability import enable_tracing
# Enable globally (default)
tracer = enable_tracing() # [!code highlight]
# Disable globally (master switch OFF)
tracer = enable_tracing(enabled=False) # [!code highlight]
```
**Important:** If `enabled=False` globally, NO agents will be traced, even if they explicitly set `tracing=True`.
## Agent-Level Control
You can opt specific agents in or out of tracing:
```python
# Opt-in (redundant if global is enabled, but good for clarity)
agent1 = create_agent(..., tracing=True) # [!code highlight]
# Opt-out (skip tracing for this agent)
agent2 = create_agent(..., tracing=False) # [!code highlight]
```
## Pool-Level Control
Pools also support the `tracing` parameter, which applies to all agents in the pool unless they have their own explicit setting.
```python
# Enable tracing for the pool
pool = create_pool(agents=[a1, a2], tracing=True) # [!code highlight]
# Disable tracing for the pool
pool = create_pool(agents=[a1, a2], tracing=False) # [!code highlight]
```
**Note:** An agent's explicit `tracing` setting always overrides the pool's setting.
## What's Next?
**[Tracing Storage](/docs/tracing-and-observability/tracing-storage)**
Configure persistent storage for your traces using SQLite, PostgreSQL, or Redis.
# Cost Tracking
Cost Tracking
Track and optimize LLM API costs with automatic token counting and pricing
Peargent automatically tracks token usage and calculates costs for all LLM API calls. This helps you monitor spending, optimize prompts, and control costs in production.
## How Cost Tracking Works
Cost tracking is automatic when tracing is enabled. Peargent counts tokens using `tiktoken` and calculates costs based on the model's pricing.
```python
from peargent import create_agent
from peargent.observability import enable_tracing
# Enable tracing to start cost tracking
tracer = enable_tracing()
agent = create_agent(..., tracing=True)
result = agent.run("What is 2+2?")
# Check costs
trace = tracer.list_traces()[-1]
print(f"Total cost: ${trace.total_cost:.6f}") # [!code highlight]
```
## Custom Model Pricing
Add pricing for custom or new models:
```python
tracer = enable_tracing()
# Add custom pricing (prices per million tokens)
tracer.add_custom_pricing(
model="my-custom-model",
prompt_price=1.50, # $1.50 per million tokens
completion_price=3.00 # $3.00 per million tokens
)
```
## Cost Calculation Formula
Costs are calculated using this formula:
```python
prompt_cost = (prompt_tokens / 1,000,000) * prompt_price
completion_cost = (completion_tokens / 1,000,000) * completion_price
total_cost = prompt_cost + completion_cost
```
## Viewing Cost Information
You can view costs per trace or get aggregate statistics:
```python
# Per-trace costs
for trace in tracer.list_traces():
print(f"{trace.agent_name}: ${trace.total_cost:.6f} ({trace.total_tokens} tokens)") # [!code highlight]
# Summary statistics
tracer.print_summary()
```
## Best Practices
1. **Enable Tracing in Production**: Always track costs in live environments.
2. **Monitor Daily**: Use `tracer.print_summary()` to check daily spend.
3. **Set Alerts**: Implement budget alerts for cost spikes.
4. **Optimize Prompts**: Reduce token usage to lower costs.
5. **Use Cheaper Models**: Use smaller models (e.g., `gemini-2.0-flash`) for simple tasks.
## What's Next?
**[Tracing Storage](/docs/tracing-and-observability/tracing-storage)**
Set up persistent trace storage with SQLite, PostgreSQL, or Redis for long-term cost analysis and reporting.
# Tracing and Observability
Tracing and Observability
Monitor agent performance, track costs, and debug with comprehensive tracing
Tracing gives you complete visibility into what your agents are doing. Track LLM calls, tool executions, token usage, API costs, and performance metrics in real-time.
\| Tracing is fully optional and adds minimal overhead (usually \<10ms), making it safe for production.
## Why Tracing?
Production AI applications need observability:
* **Cost Control** - Track token usage and API costs per request
* **Performance Monitoring** - Measure latency and identify bottlenecks
* **Debugging** - See exactly what happened when something fails
* **Usage Analytics** - Understand how agents and tools are being used
## Quick Start
Enable tracing in one line:
```python
from peargent import create_agent
from peargent.observability import enable_tracing
from peargent.models import openai
# Enable tracing
tracer = enable_tracing() # [!code highlight]
# Create agent with tracing enabled
agent = create_agent(
name="Assistant",
description="Helpful assistant",
persona="You are helpful",
model=openai("gpt-4o"),
tracing=True
)
# Run agent - traces are automatically captured
result = agent.run("What is 2+2?")
# View traces
tracer.print_summary()
```
**Output:**
```text
TRACE SUMMARY
Total Traces: 1
Total Tokens: 127
Total Cost: $0.000082
```
## What Gets Traced?
Every agent execution creates a **Trace** containing multiple **Spans**:
### **Trace**
Represents a full agent execution. Includes:
* **ID** – Unique identifier
* **Agent** – Which agent ran
* **Tokens & Cost** – Total usage and API cost
* **Duration** – Total time taken
### **Spans**
Individual operations inside a trace:
* **LLM Call** – Model, tokens, cost, latency
* **Tool Execution** – Tool name, inputs, outputs, duration
* **Agent Logic** – Reasoning steps
## Storage Options
Peargent supports multiple storage backends including **In-Memory** (default), **SQLite**, **PostgreSQL**, **Redis**, and **File-based**.
See **[Tracing Storage](/docs/tracing-and-observability/tracing-storage)** for detailed setup and configuration.
## Viewing Traces
List all traces or print a summary:
```python
# List traces
traces = tracer.list_traces() # [!code highlight]
for trace in traces:
print(f"Trace {trace.id}: {trace.total_cost:.6f}")
# Print summary
tracer.print_summary() # [!code highlight]
```
## What's Next?
**[Cost Tracking](/docs/tracing-and-observability/cost-tracking)**
Learn about model pricing, cost calculation, and optimization strategies.
**[Tracing Storage](/docs/tracing-and-observability/tracing-storage)**
Set up SQLite, PostgreSQL, or Redis for persistent trace storage.
# Session and User Context
Session and User Context
Tag traces with session and user IDs for multi-user applications
When building multi-user applications, you often need to track which user or session triggered each agent execution. Peargent provides context functions that automatically tag all traces with session and user IDs using **thread-local storage**.
## Context API
Use these functions to manage context for the current thread:
```python
from peargent.observability import (
set_session_id, get_session_id,
set_user_id, get_user_id,
clear_context
)
# Set context
set_session_id("session_123") # [!code highlight]
set_user_id("user_456") # [!code highlight]
# Get context
print(f"Session: {get_session_id()}") # [!code highlight]
print(f"User: {get_user_id()}") # [!code highlight]
# Clear context (important for thread reuse)
clear_context()
```
## Web Application Integration
### Flask Middleware
Automatically set context from request headers or session:
```python
@app.before_request
def before_request():
set_session_id(session.get('session_id'))
set_user_id(request.headers.get('X-User-ID'))
@app.after_request
def after_request(response):
clear_context()
return response
```
### FastAPI Middleware
Use middleware to handle context for async requests:
```python
@app.middleware("http")
async def add_context(request: Request, call_next):
set_session_id(request.headers.get("X-Session-ID"))
set_user_id(request.headers.get("X-User-ID"))
response = await call_next(request)
clear_context()
return response
```
## Filtering by Context
Once tagged, you can filter traces by session or user:
```python
# Filter by session
session_traces = tracer.list_traces(session_id="session_123") # [!code highlight]
# Filter by user
user_traces = tracer.list_traces(user_id="alice") # [!code highlight]
```
## What's Next?
**[Accessing Traces](/docs/tracing-and-observability/accessing-traces)**
Learn how to retrieve, filter, and analyze your trace data programmatically.
# Tracing Storage
Tracing Storage
Persist traces to SQLite or PostgreSQL for production-grade observability
Tracing storage persists traces beyond program execution, enabling historical analysis, cost reporting, and production monitoring.
## Storage Options
Peargent provides five storage backends for traces:
### In-Memory (Default)
**Best for:** Development and testing. Zero setup, but data is lost on exit.
```python
from peargent.observability import enable_tracing
tracer = enable_tracing() # [!code highlight]
```
### File-Based Storage
**Best for:** Small-scale apps and debugging. Simple JSON files.
```python
from peargent.storage import File
tracer = enable_tracing(store_type=File(storage_dir="./traces")) # [!code highlight]
```
### SQLite Storage
**Best for:** Local production. Single-file database, fast queries.
```python
from peargent.storage import Sqlite
tracer = enable_tracing(store_type=Sqlite(connection_string="sqlite:///./traces.db")) # [!code highlight]
```
### PostgreSQL Storage
**Best for:** Multi-server production. Scalable and powerful.
```python
from peargent.storage import Postgresql
tracer = enable_tracing(
store_type=Postgresql(connection_string="postgresql://user:pass@localhost/dbname") # [!code highlight]
)
```
### Redis Storage
**Best for:** High-speed caching and distributed systems.
```python
from peargent.storage import Redis
tracer = enable_tracing(
store_type=Redis(host="localhost", port=6379, key_prefix="my_app") # [!code highlight]
)
```
## Storage Comparison
| Feature | In-Memory | File | SQLite | PostgreSQL | Redis |
| ----------------- | --------- | ------ | ---------------- | --------------- | --------------- |
| Persistence | ❌ | ✅ | ✅ | ✅ | ⚠️ Optional |
| Query Performance | Fastest | Slow | Fast | Fast | Fastest |
| Concurrent Access | ❌ | ❌ | ⚠️ Limited | ✅ Excellent | ✅ Excellent |
| Production Ready | ❌ | ❌ | ⚠️ Single-server | ✅ Yes | ✅ Yes |
| Setup Required | ❌ None | ❌ None | ❌ None | ✅ Server needed | ✅ Server needed |
## Custom Table Names
You can customize table names to avoid conflicts with your existing schema.
### SQLite & PostgreSQL
Use `table_prefix` to namespace your tables (creates `{prefix}_traces` and `{prefix}_spans`).
```python
tracer = enable_tracing(
store_type=Sqlite(
connection_string="sqlite:///./traces.db",
table_prefix="my_app"
)
)
```
### Redis
Use `key_prefix` to namespace your Redis keys (creates `{prefix}:traces:*`).
```python
tracer = enable_tracing(
store_type=Redis(
host="localhost",
port=6379,
key_prefix="my_app"
)
)
```
## What's Next?
**[Session and User Context](/docs/tracing-and-observability/session-context)**
Learn how to tag traces with user and session IDs for better organization and analysis.