agentl MCP server

<!-- mcp-name: io.github.arimayuji/agentl -->

Website · Getting started · Connectors · PyPI · Roadmap · Discussions

Agentic ETL in YAML. Read files, APIs and databases, let an LLM classify or extract each record against a schema, check the data, and deliver it. Per-record cache, budgets, retries and a live dashboard come built in.

halfway only pays for what did not finish (before and after).

• Pay once per record: a re-run of unchanged data makes no model calls, and a job that failed

Kafka, SQS, webhooks, and an MCP server for Claude Code and Cursor.

• Data you can trust: check steps with fixed rules and an agent review before delivery.
• Works where your data is: CSV, Excel, PDF, SQL, Postgres, S3, SharePoint, Google Sheets,

evals against plain Python, including where it loses.

• Honest numbers: benchmarks and

Status: beta (v0.9). Minor releases may still change the API; every change is in the changelog.

For AI agents: the docs are at agentl.com.br/llms.txt, and every docs page has a raw Markdown copy (/index.md).

Install

pip install agentl-etl                 # core: CSV, JSON, HTTP, SQLite, runner, dashboard
pip install 'agentl-etl[office,pdf]'   # Excel, Word, PDF
pip install 'agentl-etl[anthropic]'    # Claude as the LLM provider
pip install 'agentl-etl[all]'

The distribution is named agentl-etl on PyPI; the import and the CLI are both agentl.

A pipeline

name: invoices
llm:
  provider: anthropic          # or "echo" to run offline
  model: claude-opus-5-5

steps:
  - id: invoices
    use: csv.read
    with: { path: data/invoices.csv }

  - id: customers
    use: json.read
    with: { path: data/customers.json }

  - id: matched
    use: join
    input: [invoices, customers]
    with: { left: invoices, right: customers, on: tax_id, how: left }

  - id: classify
    agent: Classify the expense category and flag unknown clients for review.
    input: matched
    output: { number: str, customer: str?, category: str?, needs_review: bool? }

  - id: save
    use: sqlite.write
    input: classify
    with: { db: out/invoices.db, table: invoices }

notify:
  channels:
    slack: { webhook: "${SLACK_WEBHOOK}" }
  rules:
    - on: run.failed
      to: [slack]
      summary: agent             # the LLM explains the failure in 3 lines
    - on: item.rejected
      when: "rate > 0.05"
      to: [slack]

Shortened. The full README is on GitHub.