Liquid

Connect your AI agent to anything — with no connector to write or maintain.

Point Liquid at a URL or a database and it works out the interface for you: discovers its shape, maps it to the fields you asked for, and handles auth, pagination and normalization — typed records, no client code. When the upstream drifts, it re-maps and keeps going. The same small API — fetch · query · write · sense — reaches web APIs, databases, other agents (MCP/A2A), email, and even IoT and industrial systems (MQTT, Modbus, OPC UA, BACnet). An LLM does the learning at setup (and on drift); the data path itself makes no model call.

---

What an agent can reach through Liquid

One agent-facing API (fetch · query · write · sense) over everything an agent might need to touch — Liquid figures out how to talk to it so the agent doesn't have to. It's the agent's senses and hands: fetch/query probe, sense perceives a live event stream, write acts on the world.

• Web APIs & messaging — REST/JSON, GraphQL, SOAP/WSDL, gRPC, WebSocket, SSE/NDJSON streams, MQTT (IoT pub/sub — subscribe to sense, publish to act)
• Email — IMAP/SMTP (any provider, app-password or OAuth2 XOAUTH2) and the Gmail API (OAuth2): read a mailbox, sense new mail as it arrives, and send
• Industrial / OT — Modbus (PLCs, sensors) and OPC UA (Industry-4.0 nodes, native subscriptions) for the factory floor; BACnet for buildings (HVAC/BMS) — read, write, and sense
• Android devices — phones, TV boxes, kiosks via ADB: sense logcat, read shell, act with input/am
• Other agents & tools — any MCP server, A2A agents, ChatGPT-plugin manifests
• Databases — Postgres (+ pgvector), MySQL/MariaDB, SQLite, DuckDB, SQL Server, Neo4j (graph), MongoDB (documents), Redis (key-value)
• People, places & things — a human, a home, or a car as a node via connectors: Telegram (perceive messages, send replies), Home Assistant (perceive a whole smart home's state changes, act via call_service — lights, locks, media), and Smartcar (perceive a connected vehicle across ~30 brands — location/battery/fuel — and act: lock/unlock, charge)

Point it at a https://… endpoint, a postgres://… / mongodb://… / redis://… DSN, a grpc://… target, or another MCP server — discovery identifies the interface, learns its shape, and hands your agent typed records. The same fetch/query/write works regardless of what's underneath. No per-service connector to hand-write; the integration maintains itself when the upstream changes.

# A web API it has never seen — no spec, no connector, no auth
adapter = await liquid.get_or_create(
    "https://api.openbrewerydb.org/v1/breweries",
    target_model={"name": "str", "city": "str", "country": "str"},
    auto_approve=True,
)
breweries = await liquid.fetch(adapter)            # typed records

# A database is just another interface — same API, and it writes too
db = await liquid.get_or_create("postgresql://reader@host/shop",
                                target_model={"id": "int", "email": "str"},
                                auto_approve=True)
orders = await liquid.fetch(db, "/public/orders")
await liquid.write(db, "/public/orders", op="insert",
                   values={"email": "a@b.com", "total_cents": 9900},
                   allow_write=True)               # opt-in; mutates the store

You hand-write no connector and no schema: an LLM learns the interface once at setup (databases introspect themselves and skip even that), and the integration repairs itself when the upstream drifts. The runtime is plain deterministic transport — predictable cost, reproducible behavior, nothing to babysit.

Built for the constraints real agents hit

Reaching everything is half of it. The other half is that agents pay for every token, get confused by inconsistent shapes, and can't parse error prose. Liquid answers each with a concrete primitive — all shipped, all on PyPI.

Context-budget control

# Search / aggregate server-side instead of fetch-then-filter — 10-100x fewer tokens
orders = await liquid.search(adapter, "/orders",
    where={"total_cents": {"$gt": 10000}, "status": "paid"}, limit=20)

stats = await liquid.aggregate(adapter, "/orders",
    group_by="status", agg={"total_cents": "sum", "id": "count"})

hits = await liquid.text_search(adapter, "/tickets", "shipping delay")  # BM25-lite

data = await liquid.fetch(adapter, "/orders", max_tokens=2000)      # budget cap
data = await liquid.fetch(adapter, "/customers", verbosity="terse") # id + 1-2 fields

Cross-source normalization

liquid = Liquid(..., normalize_output=True)
# Stripe {amount:1000,currency:"usd"} · PayPal {value:"10.00",currency_code:"USD"}
#   → Money(amount_cents=1000, currency="USD", amount_decimal=Decimal("10.00"))

Timestamps (Unix / ISO 8601 / RFC 2822) collapse to UTC datetime; pagination envelopes ({data:[…]} / {results:[…]} / Link headers) flatten; ID fields normalize across id / _id / uuid / *_id.

Canonical intents — one mental model across services

await liquid.execute_intent(adapter, "charge_customer",
    {"customer_id": "cus_xyz", "amount_cents": 9999, "currency": "USD"})
# Same intent on Stripe / Braintree / Square / Adyen — 71 canonical intents

Structured recovery — agents self-heal without parsing text

try:
    await liquid.fetch(adapter, "/orders")
except LiquidError as e:
    if e.recovery and e.recovery.next_action:
        await agent.call_tool(e.recovery.next_action.tool, e.recovery.next_action.args)

Every error carries a Recovery with next_action: ToolCall, retry_safe, and retry_after_seconds. 401 → store_credentials. 404/410 → repair_adapter. 429 → retry after the given delay. And when the upstream's schema drifts, adapters self-heal (repair_adapter) — the agent keeps working.

Predictable cost — know before you call

est = await liquid.estimate_fetch(adapter, "/orders")
# FetchEstimate(expected_items=250, expected_tokens=52_000, confidence="high", …)
if est.expected_tokens < my_budget:
    data = await liquid.fetch(adapter, "/orders")

Tools emitted by to_tools() carry a metadata block (cost_credits, typical_latency_ms, cached, idempotent, side_effects, related_tools) so the agent can reason about which tool to pick — and ambient tools (liquid_check_quota, liquid_list_adapters, …) let it ask about state instead of memorizing it.

---

Measured impact

Deterministic benchmarks on realistic agent tasks (500-order, 200-ticket fixtures, mocked HTTP) — reproducible via python -m benchmarks.run:

| Task | Metric | Baseline | With Liquid | Delta | |---|---|---:|---:|---:| | Find 10 orders over $100 | tokens | 75,482 | 1,519 | −98% | | Revenue by status (aggregate) | tokens | 75,482 | 115 | −100% | | Fetch customer (id+email only) | tokens | 424 | 12 | −97% | | Recover from 401 | structured next_action | no | yes | — | | Find the shipping ticket | tokens | 14,588 | 154 | −99% | | Stripe↔PayPal consistency | field overlap | 0.11 | 1.00 | +9× | | Skip wasted call via estimate | tokens | 14,943 | 0 | −100% | | max_tokens=2000 budget cap | tokens | 14,943 | 1,999 | −87% |

Full methodology + per-task breakdown: benchmarks/RESULTS.md.

Install

pip install liquid-api                 # core + bundled MCP server (the `liquid-mcp` command)
pip install 'liquid-api[discovery]'    # + an LLM for discovering spec-less REST APIs & field mapping

Do you need an LLM extra? Self-describing interfaces — OpenAPI, GraphQL, gRPC, MCP, A2A, WSDL — and all databases (introspection) discover with no LLM, and the whole runtime (fetch/query/write/sense) never calls a model. You only need an LLM backend to discover a REST API that has no machine-readable spec (heuristic + LLM) and to map its fields. [discovery] pulls LiteLLM, which reaches OpenAI / Gemini / Anthropic / local / 100+ providers; or pick one directly:

pip install 'liquid-api[gemini]'     # Google Gemini   (or [anthropic]; OpenAI/local work with no extra via base_url)
pip install 'liquid-api[grpc]'       # gRPC transport (reflection)
pip install 'liquid-api[ws]'         # WebSocket transport
pip install 'liquid-api[pg]'         # Postgres / pgvector (asyncpg)
pip install 'liquid-api[mysql]'      # MySQL / MariaDB (aiomysql); SQLite needs no extra
pip install 'liquid-api[neo4j]'      # Neo4j graph (Bolt / Cypher)
pip install 'liquid-api[duckdb]'     # DuckDB (embedded analytics)
pip install 'liquid-api[mssql]'      # SQL Server (ODBC; needs a system ODBC driver)
pip install 'liquid-api[mongodb]'    # MongoDB (collections as endpoints)
pip install 'liquid-api[redis]'      # Redis (keyspace namespaces as endpoints)
pip install 'liquid-api[mqtt]'       # MQTT (IoT pub/sub)
pip install 'liquid-api[modbus]'     # Modbus (industrial registers)
pip install 'liquid-api[opcua]'      # OPC UA (Industry-4.0 nodes + subscriptions)
pip install 'liquid-api[bacnet]'     # BACnet (building automation; ADB needs the system `adb` binary)
# Framework integration (LangChain / OpenAI / Anthropic / MCP) is built in — no extra package.

The core is dependency-free — every backend's library is an optional extra, imported only when used.

See it work — live, no pre-config

Point Liquid at an API it has never seen (no adapter, no OpenAPI spec, no auth) and get typed records back — you write no connector; discovery + mapping is the only place a model runs. Runnable end to end via examples/live_quickstart.py:

Connecting to an API Liquid has never seen:
  https://api.openbrewerydb.org/v1/breweries

  discovery method : rest_heuristic
  mapped fields    : ['name', 'city', 'state', 'country']
  LLM calls so far : 2  (discovery + mapping)

fetch() -> 50 typed records; first 3:
   {'name': '(405) Brewing Co', 'city': 'Norman', 'state': 'Oklahoma', 'country': 'United States'}
   {'name': '(512) Brewing Co', 'city': 'Austin', 'state': 'Texas', 'country': 'United States'}
   {'name': '1 of Us Brewing Company', 'city': 'Mount Pleasant', 'state': 'Wisconsin', 'country': 'United States'}

  LLM calls during fetch : 0
  LLM calls on 2nd fetch : 0

You wrote no connector, no schema, no auth glue — Liquid learned the interface for you, and will re-learn it if it changes. That's the point: integrations you don't build or babysit.

Run as an MCP server (open source, self-hosted)

Expose the engine to any MCP client (Claude Desktop, Cursor, Claude Code) — it runs in your own process, no cloud, no account, no lock-in:

One-click in Cursor (the button writes the server into your mcp.json; add your OPENAI_API_KEY in Cursor's MCP settings afterward). Or set it up manually:

pip install liquid-api
export OPENAI_API_KEY=sk-...        # or GEMINI_API_KEY / ANTHROPIC_API_KEY,
                                    # or OPENAI_BASE_URL=http://localhost:11434/v1 for local (Ollama/vLLM)
liquid-mcp                          # or: python -m liquid.mcp_server

Zero-install with uvx (the liquid-mcp package makes the command run by name) — Claude Code:

claude mcp add liquid --scope user -e OPENAI_API_KEY=sk-... -- uvx liquid-mc