Agents
Recipes for building agents with run_agent. For per-toolkit reference (constructor params, available methods), see Tools. For the full run_agent reference (parameters, return type), see run_agent.
When to use run_agent vs prompt
Use prompt when... | Use run_agent when... |
|---|---|
| You need a single answer or extraction | The task requires multiple steps or decisions |
| Input and output are well-defined | The agent needs to read files, query databases, or call APIs |
| No tools or external actions needed | Each step depends on the result of the previous one |
You want structured output with format | The task involves unstructured content (emails, documents, screenshots) |
Basic agent
from abstra.ai import run_agent
from abstra.tasks import get_trigger_task
task = get_trigger_task()
result = run_agent(
prompt=f"""
You are a customer support agent.
Customer: {task.payload.get("customer_name")}
Message: {task.payload.get("message")}
Classify this request and propose a response.
""",
max_steps=20,
)
print(result)
Custom tools
Any Python function can be a tool. Use type hints and docstrings so the agent knows when and how to call it.
from abstra.ai import run_agent
def lookup_order(order_id: str) -> str:
"""Returns the status of a given order by its ID."""
orders = {
"123": "In transit - arriving March 15",
"456": "Delivered on March 10",
}
return orders.get(order_id, "Order not found")
def calculate_shipping(weight_kg: float, destination: str) -> str:
"""Calculates shipping cost based on weight and destination."""
base_cost = weight_kg * 5.0
if destination == "international":
base_cost *= 2.5
return f"R${base_cost:.2f}"
result = run_agent(
prompt="Check orders 123 and 456. Then calculate shipping for a 3kg international package.",
tools=[lookup_order, calculate_shipping],
)
print(result)
Tips:
- Tool names are derived from Python function/method names — pick descriptive names.
- The docstring becomes the tool description the agent sees.
- Type hints define the JSON schema for arguments.
Routing tasks with custom tools
Use plain functions to route the task to different downstream stages based on the agent's reasoning.
from abstra.ai import run_agent, FilesTools
from abstra.tasks import get_trigger_task, send_task
task = get_trigger_task()
def approve(summary: str) -> str:
"""Mark the document as approved and forward it with a summary."""
send_task("approved", {"summary": summary})
return "Document approved and forwarded."
def reject(missing_fields: str) -> str:
"""Mark the document as rejected with the list of missing fields."""
send_task("rejected", {"missing_fields": missing_fields})
return "Document rejected and forwarded."
run_agent(
prompt=f"""
Review the following document submission:
- Type: {task.payload.get("document_type")}
- Submitted by: {task.payload.get("submitted_by")}
The document is stored at: {task.payload.get("file_path")}
Read the document and determine if it is complete. Then:
- If complete, call approve() with the document summary
- If incomplete, call reject() with the missing fields
""",
tools=[
FilesTools(actions=["read_text"], globs=[task.payload.get("file_path", "*")]),
approve,
reject,
],
)
Multimodal: images and PDFs in the prompt
Pass files directly in the prompt list. Images and PDFs are forwarded to the LLM as multimodal input.
from abstra.ai import run_agent, TablesTools
from abstra.tasks import get_trigger_task
task = get_trigger_task()
run_agent(
prompt=[
"""Extract the following information from the receipt:
- Store name
- Date
- Total amount
- List of items with prices
Insert the data into the `receipts` table.""",
task.payload.get("receipt_image"),
],
tools=[TablesTools(method=["insert"], table="receipts")],
)
Combining multiple toolkits
Mix built-in toolkits and custom functions. The agent decides which to use based on the prompt.
from abstra.ai import run_agent, TablesTools, FilesTools, ConnectorsTools
run_agent(
prompt="""
Read the CSV file with customer complaints, categorize each one,
insert the results into the complaints table, and send a Slack summary.
""",
tools=[
FilesTools(actions=["read_text"], globs=["complaints.csv"]),
TablesTools(method=["insert"], table="complaints"),
ConnectorsTools(connection="slack", action="send_message"),
],
)
End-to-end: invoice processing pipeline
A complete agent that receives invoices, extracts data, validates totals, stores rows, and notifies via Slack — combining four kinds of tools.
from abstra.ai import run_agent, TablesTools, ConnectorsTools
from abstra.tasks import get_trigger_task, send_task
task = get_trigger_task()
invoice_path = task.payload.get("file_path")
def flag_for_review(reason: str) -> str:
"""Flag the invoice for manual review with the given reason."""
send_task("manual_review", {
"file_path": invoice_path,
"reason": reason,
})
return "Invoice flagged for manual review."
def mark_as_processed(invoice_data: dict) -> str:
"""Mark the invoice as successfully processed."""
send_task("processed", {
"file_path": invoice_path,
**invoice_data,
})
return "Invoice marked as processed."
run_agent(
prompt=[
f"""
You are an invoice processing agent. Analyze the attached invoice and:
1. Extract: vendor name, CNPJ, invoice number, date, line items, and total amount
2. Validate: check that line items sum to the total (within 1% tolerance)
3. Store the extracted data in the `invoices` table
4. If valid, call mark_as_processed with the extracted data
5. If there are discrepancies, call flag_for_review with the reason
6. Send a summary to the #finance Slack channel
""",
invoice_path,
],
tools=[
TablesTools(method=["insert", "select"], table="invoices"),
ConnectorsTools(connection="slack", action="send_message"),
flag_for_review,
mark_as_processed,
],
max_steps=15,
)
End-to-end: browser automation with debugging
Combines BrowserTools with the optional listen_network, listen_console, and listen_websocket flags for full visibility into a flaky page.
from abstra.ai import run_agent, BrowserTools
result = run_agent(
prompt="""
Open https://app.example.com/dashboard, log in with the credentials in the prompt,
then click the 'Export' button. If the export fails, capture any failed network calls,
console errors, and WebSocket frames so we can diagnose the issue.
Credentials: user@test.com / pass123
""",
tools=[
BrowserTools(
url="https://app.example.com",
listen_network=True,
listen_console=True,
listen_websocket=True,
),
],
max_steps=15,
)
print(result)
Tips
- Start small: Begin with specific, restricted toolsets rather than giving broad access.
- Use
max_steps: Keep it conservative (10-20) for focused tasks. The default of 30 is generous. - Clear prompts: Be explicit about the objective, constraints, and expected output.
- Type hints + docstrings: Always add them to custom tools — the agent reads them to decide which tool to use.
- Scope access: Prefer
table="orders"over unrestricted access,globs=["data/*"]over"*". - Use
wherefor multi-tenancy: Passwhere={"tenant_id": id}to scope all database operations. - Browser: prefer
*_element(page_id, index, ...)over*(page_id, selector, ...)— indices come from the latestget_page_summaryand survive selector churn.