Skip to main content

Agents

Agents extend the AI SDK beyond simple prompt-response. Use run_agent when you need multi-step reasoning, tool use, or autonomous decision-making inside any stage.

When to use run_agent vs prompt​

Use prompt when...Use run_agent when...
You need a single answer or extractionThe task requires multiple steps or decisions
Input and output are well-definedThe agent needs to read files, query databases, or call APIs
No tools or external actions neededEach step depends on the result of the previous one
You want structured output with formatThe 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)

Agent with database access​

Use TablesTools to let the agent query and modify your Abstra Tables.

from abstra.ai import run_agent, TablesTools
from abstra.tasks import get_trigger_task

task = get_trigger_task()

run_agent(
    prompt=f"""
You received a support ticket:

- From: {task.payload.get("customer_email")}
- Subject: {task.payload.get("subject")}
- Message: {task.payload.get("message")}

Categorize this ticket by inserting a row into the `support_tickets` table with:
- email: the customer's email
- category: one of "billing", "technical", "general"
- priority: one of "low", "medium", "high"
- summary: a one-sentence summary of the issue
""",
    tools=[TablesTools(method=["insert"], table="support_tickets")],
)

Read-only queries​

from abstra.ai import run_agent, TablesTools

result = run_agent(
    prompt="List all open tickets and summarize the most common categories.",
    tools=[TablesTools(method=["select"], table="tickets")],
)

Multi-table with tenant scoping​

from abstra.ai import run_agent, TablesTools

run_agent(
    prompt="Process the pending orders for tenant abc.",
    tools=[
        TablesTools(
            method="all",
            table=["orders", "order_items"],
            where={"tenant_id": "abc"},
        )
    ],
)

Agent with file access​

Use FilesTools to let the agent read, write, and manage files in your project.

from abstra.ai import run_agent, FilesTools

run_agent(
    prompt="Read all files in data/ and create a summary report.",
    tools=[FilesTools(actions=["read_text", "list"], globs=["data/*"])],
)

Parse and transform files​

from abstra.ai import run_agent, FilesTools

run_agent(
    prompt="Parse input.csv, clean the data, and write the result to output.csv.",
    tools=[FilesTools(actions=["read_text", "write_text"], globs=["*.csv"])],
)

Agent with connectors​

Use ConnectorsTools to let the agent call external services like Slack, Stripe, or email.

from abstra.ai import run_agent, ConnectorsTools
from abstra.tasks import get_trigger_task

task = get_trigger_task()

run_agent(
    prompt=f"""
A new order was placed:
- Customer: {task.payload.get("customer_name")}
- Total: R${task.payload.get("total")}
- Items: {len(task.payload.get("items", []))} items

Send a summary to the #orders Slack channel.
""",
    tools=[ConnectorsTools(connection="slack", action="send_message")],
)

Agent with browser automation​

Use BrowserTools to give the agent a headless Chromium browser for web scraping, form filling, and screenshots.

from abstra.ai import run_agent, BrowserTools

result = run_agent(
    prompt="""
Go to https://news.ycombinator.com and extract the top 5 stories.
For each story, get the title, URL, and point count.
""",
    tools=[BrowserTools(url="https://news.ycombinator.com")],
    max_steps=10,
)

Login and screenshot​

from abstra.ai import run_agent, BrowserTools

run_agent(
    prompt="""
Go to https://app.example.com/login, fill in user@test.com and password123,
click Login, then take a screenshot of the dashboard.
""",
    tools=[BrowserTools(url="https://app.example.com")],
)

Debug with network and console tracking​

from abstra.ai import run_agent, BrowserTools

run_agent(
    prompt="Navigate to the page and check for any failed API calls or console errors.",
    tools=[BrowserTools(
        url="https://app.example.com",
        listen_network=True,
        listen_console=True,
    )],
)

Agent with 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)

Agent with custom tools and workflow transitions​

Use custom functions to route tasks to different 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,
    ],
)

Agent with images and PDFs​

Pass files directly in the prompt list. Images and PDFs are automatically processed by the LLM.

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 tools 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"),
    ],
)

Real-world example: Invoice processing pipeline​

A complete agent that receives invoices, extracts data, validates, stores in database, and notifies via Slack.

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,
)

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 where for multi-tenancy: Pass where={"tenant_id": id} to scope all database operations.