A general-purpose engine for large-scale, repeated information gathering by a fleet of workers — plain fetchers, scrapers, or LLM agents. You register things to check and how often; the engine schedules what is due, leases work out in isolated batches, records what came back, detects change by content hash, backs off on failures, and keeps an auditable ledger of every write. One SQLite file is the entire system. Python standard library only.

It came out of a production system where a mixed fleet of LLM agents and plain fetchers tracks a few thousand sources on daily and weekly cadences. This is a clean rewrite of the part that generalizes: the bookkeeping. The engine never interprets what a source is — the route field means nothing to it and everything to your worker — which is why the same engine can run very different fleets.

• Sales prospecting. Register target companies' careers pages, press pages, and funding announcements. A changed hash on a careers page is a hiring signal; your worker (or an LLM agent reading the diff) turns it into a lead with a timestamp and a receipt.
• Competitor and price watch. Pricing pages, feature pages, changelogs, app-store listings — checked on cadence, with changed flags you can alert on and history you can chart.
• Regulatory and compliance tracking. Agency pages, statute databases, published guidance. The hash ledger doubles as evidence of when a change appeared and when you saw it.
• Research corpus upkeep. Datasets, preprint feeds, documentation sets that must stay current. Re-ingest is safe because everything is idempotent by hash. Research-tool example: alphaXiv.
• Marketplace and listing monitoring. Job boards, real-estate listings, auction lots — anything where being first to notice a change is the point.

In every case the engine's contribution is the same: which of your thousands of sources are due right now, what actually changed, what is failing and backing off, and proof of all of it.

Five tables (see agentfleet/schema.sql):

• source — a thing to check: a route, a cadence, a next_due_at, an error streak
• action — one unit of work; states QUEUED → LEASED → RUNNING → DONE | FAILED, requeued while attempts remain
• batch — a lease of actions to one worker, with a write-scope directory only that worker touches
• observation — one check's result: content sha256, bytes, timings, receipt path, a changed flag
• change_log — every canonical write, stamped with an actor and a reason

Three rules carry the design. Workers never write canonical state; they drop result files in their write scope and the parent folds them in during reconcile, so a crashed or misbehaving worker can corrupt at most its own scratch directory. Change detection is a content hash, so any run can be repeated safely. Every canonical write is attributed, which matters once several agents share one database.

Scheduling: success sets next_due_at = now + cadence; failure sets next_due_at = now + cadence × min(2^streak, 16) and requeues up to 3 attempts.

python -m agentfleet.cli add-source --db fleet.db acme-careers https://example.com/careers --cadence 3600
python -m agentfleet.cli sweep --db fleet.db
python -m agentfleet.cli status --db fleet.db
python -m agentfleet.cli stats --db fleet.db

sweep queues due sources, leases batches, fetches each route over HTTP, and reconciles. stats prints per-source change rate, error rate, median and p95 latency, and fleet totals. python examples/demo.py runs the whole loop on local files with no network. Tests: python -m unittest discover -s tests -v.

Use TypeSafe's direct API to classify fetched text or explicit structured state. No SDK, gateway, database migration or key is needed for normal fleet operation. Jev is disabled unless requested.

Provide TYPESAFE_API_KEY through your environment or secret manager; never put a key in source routes, input JSON, result files or Git. Then try the synthetic example:

python -m agentfleet.cli jev-evaluate examples/jev_request.json

The request file contains state and questions. Supported questions are choice, noul (yes/no probability) and score. Output contains the resolved model, typed answers and token usage. This command does not open or modify a fleet database. The default model is pinned to jev-1.13.0; override it with --model. --timeout defaults to 30 seconds.

To evaluate each successfully fetched source during a sweep:

python -m agentfleet.cli sweep --db fleet.db --scopes ./scopes \
  --jev-questions examples/jev_questions.json

This opt-in sends fetched UTF-8 text and your rubric to TypeSafe. Use it only for data you are authorized to share; extract and sanitize documents before classification when needed. Binary input fails explicitly, and oversized requests are rejected rather than silently truncated. Source text is untrusted evidence; write rubrics that preserve uncertainty for review.

Each worker result JSON gains evaluation (model, answers, usage), fetch_elapsed_ms, and evaluation_elapsed_ms; elapsed_ms covers both phases. The .payload and sha256 still describe the original source content, so changing model probabilities do not create false source-change alerts. Evaluation data is available through the observation's receipt_path, not new database columns. A valid classification is not an approval, rejection or proof of complete source coverage; the parent decides how to use it.

API/validation failures fail the action and retain any fetched payload for inspection. Existing fleet retries/backoff still apply. The client itself makes one request, rejects redirects, limits request/response size to 1 MiB and never falls back to another provider. TypeSafe's token limits still apply below those byte limits. The CLI checks credentials and question configuration before queueing work. Sweep overrides: --jev-model and --jev-timeout.

Python workers can use the same integration:

from agentfleet import run_batch
from agentfleet.jev import JevClient

client = JevClient()  # TYPESAFE_API_KEY; no credentials in the result
result = client.evaluate({"text": "A public changelog"}, questions)
run_batch(conn, batch, fetcher, evaluator=client.text_evaluator(questions))

run_batch also accepts any evaluator(content_bytes) -> JSON dict. Evaluation is synchronous; the caller controls worker concurrency and API rate limits. Keep result scopes private if they contain sensitive source data. Error messages omit provider response bodies and credentials.

The built-in executor is a function route -> (status, content_bytes). For an external fleet (LLM agents, containers, other machines), call lease() to get a manifest, hand each batch to a worker, and have the worker write {action_id}.json result files into its write_scope:

{"action_id": 7, "source_id": 3, "status": "ok", "sha256": "…", "bytes": 4096,
 "started_at": "2026-08-31T19:50:21Z", "finished_at": "2026-08-31T19:50:22Z", "elapsed_ms": 812.4}

Then reconcile(conn, batch_id). The engine treats a missing result file as a failure, so dead workers resolve themselves. An LLM agent makes a good worker precisely because the contract is this narrow: read the manifest, do the check however you judge best, write result files, touch nothing else. SKILL.md teaches a coding agent to operate the whole loop.

The tables are span-shaped by construction, so telemetry is an export:

python -m agentfleet.cli export-otel --db fleet.db --endpoint http://localhost:4318

maps run → trace, batch → parent span, action → child span (with source name, hash, bytes, attempt, and changed-flag attributes) and posts OTLP/HTTP JSON to any collector, including a Grafana Cloud OTLP gateway. Omit --endpoint to inspect the JSON on stdout.

Apache-2.0. Built by Dreamers Inc.