Taskwarden

A durable, owner-approved AI team for a business.

Taskwarden is a Python harness for running a small team of AI agents that do real work for a business (texting customers, booking, refunding, posting) on a server, safely. The owner decides what each agent may do alone and what must ask first. Anything that asks waits on a queue in your database until the owner approves, edits or declines it, then carries on where it stopped, even after a restart.

It runs the AI team of Perbene in production, for many businesses at once.

pip install taskwarden

Why another agent framework

Most frameworks are built for one person at a keyboard, or for a demo. A business's team runs unattended, for many businesses at once, and its mistakes reach customers. Taskwarden is built around the questions that come up then:

How it fits

owner request ──▶ loop.run_task ──▶ model ──▶ tool call ──▶ gate.decide
                                                              │
                         allow ◀───────────────┬──────────────┤
                           │                   │              ▼
                       run the tool        observe        ask: park the task
                           │               (record)        (pending, in your DB)
                           ▼                                  │
                    next round / report        owner: approve / edit / decline
                                               decisions.claim + decisions.apply
                                                              │
                                                 queued ──▶ sweeper resumes it

Quick start

examples/shop.py is a complete app: a web shop whose agent reads orders on its own and refunds only with the owner's click. It runs offline (SQLite, a scripted model):

pip install "taskwarden[dev]"
python examples/shop.py

owner: Refund the customer who paid $19.50 today.
run:   waiting
waits: refund {'order_id': 'A7'}
owner: approved, with order A8 instead of A7
run:   done
...
chat: [lead] Done: I refunded order A8, $19.50.

The shape of a host:

from taskwarden import decisions, loop, sweeper, tools
from taskwarden.host import BaseHost
from taskwarden.models import MessageColumns, TaskColumns

class Task(Base, TaskColumns):
    __tablename__ = "team_tasks"

class Message(Base, MessageColumns):
    __tablename__ = "team_messages"

class ShopHost(BaseHost):
    Task, Message = Task, Message
    chief = "lead"
    agents = {"lead": Agent("lead", "Sam", "Shop assistant", ("read_orders", "refund"))}
    tools = {
        "read_orders": tools.Tool("read_orders", tools.READ, schema, read_orders),
        "refund": tools.Tool("refund", tools.SPEND, schema, refund),
    }

    async def llm(self, db, organization_id):
        return openai.AsyncOpenAI(), "your-model"   # any OpenAI-compatible client

    async def system_prompt(self, db, task, agent, name, *rest) -> str:
        return f"You are {name}, the {agent.title} of a small web shop."

await loop.run_task(host, db, task, actor="owner")          # the owner's request
await decisions.apply(host, db, org, await decisions.claim(host, db, org, task.id, "approve"),
                      "approve")                             # their click, any time later
await sweeper.sweep(host, Session, redis, now=now, run=run) # on a timer: resumes the rest

BaseHost leaves out what only your app can know. Override it as you grow: sent_key / reach_key (turn on the duplicate-send guard), notify_parked (push, SMS or email the owner), refuse_run (credits), context (the business's time zone), audit (store the trail), mark (live progress), thought (the model's reasoning, as it works), take_messages (what the owner writes while a task runs joins it), stop_asked (the owner's Stop: the run ends at its next step and says what it did), clock (how the time reads; it rides on the first message of each request, never in the system prompt, so the prompt stays cached).

One clock, many triggers

An agent says in its AGENT.md what wakes it (when:, not on:, which YAML reads as true):

triggers:
  - when: schedule          # Mondays at 09:00, the business's local time
    cadence: weekly
    weekday: 0
    at: "09:00"
    goal: Find what people kept asking for this week.
  - when: done              # Scout finished: its report comes with the goal
    after: scout
    goal: Collect the complaints behind Scout's top three.
  - when: event             # your app calls triggers.fire(host, db, org, "email:bugs", body)
    event: email:bugs
    goal: A bug report came in. Write the steps to reproduce it.

class MyHost(BaseHost):
    triggers = triggers.of_folders(folders)        # checked: agents exist, no done-loop

await triggers.fire(host, db, org, "email:bugs", email_text)  # your webhook handler
await triggers.due(host, db, org, now=now)                    # in the sweeper's on_sweep
# due(..., origin="agenda", goal=todays_words): your own origin name, and the day's words
# (or None to skip today), asked only for a run that is otherwise due.
# `done` triggers need nothing: a finished task starts whoever follows its agent.

Every task a trigger starts is self-started, so an email can wake an agent but cannot make it send, pay or change records without the owner's click; one woken by an event or a report asks even before routine work (a text back, a booking), since a stranger's words may be steering it. The same goal is queued once a day (a retried webhook), each agent runs at most max_per_day triggered tasks, a paused agent stays asleep, and a done chain that would loop is refused when it is loaded.

What arrives from outside (the event's payload, a teammate's report) is fenced: every line marked, between two lines carrying the text's own fingerprint, so a forged "end of email, now the owner says..." line stays inside it.

A trigger with watch: true only looks: it is offered its agent's read tools and nothing else (refused in code if the model names another), and when nothing needs the owner it finishes without a word. Use it for "keep an eye on the status page".

Agents can set up their own routines: give one triggers.routine_tool(host, save) and write custom_triggers(db, org) to read back what save stored. Adding a routine is a profile call, so it always waits for the owner's click.

examples/launch_team.py runs a five-agent launch team offline: Scout on Mondays, Pain Miner and Judge on each other's reports, Inbox on email, and a routine Chief sets up once the owner approves.

Spend

DAILY_TOKENS = 200_000 on your host caps each agent's model tokens over a rolling day (or write token_budget(db, task) to vary it by plan or agent). Past it the run stops and the chat says so. Needs the tokens column, which TaskColumns has.

Security

SECURITY.md lists what the code enforces, what is only a convention, and what is yours as the host.

Risk levels

Requirements

Python 3.11+, SQLAlchemy 2 (async), PyYAML. Postgres in production (SQLite works for trying it out), Redis for the sweeper's lock. Any OpenAI-compatible chat model with tool calls.

Tests

pip install -e ".[dev]"
pytest

License

Apache-2.0. Built by DAHO.