Liquid

Liquid

将 AI Agent 连接到任何服务——无需编写或维护连接器。

只需让 Liquid 访问一个 URL 或数据库，它就会为你确定接口： 发现其结构，将其映射到你指定的字段，并处理身份验证、 分页和规范化——返回类型化记录，无需客户端代码。当上游 发生变化时，它会重新映射并继续运行。同一套小型 API——fetch · query · write · sense——可访问 Web API、数据库、其他 Agent（MCP/A2A）、电子邮件， 甚至 IoT 和工业系统（MQTT、Modbus、OPC UA、BACnet）。大模型在设置时（以及上游 发生变化时）进行学习；数据路径本身不会调用模型。

---

Agent 可以通过 Liquid 访问什么

一套面向 Agent 的 API（fetch · query · write · sense），可用于 Agent 可能需要接触的一切——Liquid 会确定如何与其交互，无需 Agent 操心。它既是 Agent 的感官也是双手：fetch/query 用于探查， sense 感知实时事件流，write 则用于影响现实世界。

• Web API 和消息传递——REST/JSON、GraphQL、SOAP/WSDL、gRPC、WebSocket、 SSE/NDJSON 流、MQTT（IoT 发布/订阅——订阅以感知，发布以执行操作）
• 电子邮件——IMAP/SMTP（任意提供商，应用专用密码或 OAuth2 XOAUTH2）以及 Gmail API（OAuth2）：读取邮箱、用 sense 感知新邮件到达，并发送邮件
• 工业 / OT——Modbus（PLC、传感器）和 OPC UA（工业 4.0 节点、 原生订阅），适用于工厂车间；BACnet 适用于建筑（HVAC/BMS）—— 读取、写入和感知
• Android 设备——通过 ADB 连接手机、电视盒子、信息亭：感知 logcat，读取 shell，使用 input/am 执行操作
• 其他 Agent 和工具——任意 MCP 服务、A2A Agent、ChatGPT 插件清单
• 数据库——Postgres（+ pgvector）、MySQL/MariaDB、SQLite、DuckDB、SQL Server、 Neo4j（图数据库）、MongoDB（文档数据库）、Redis（键值数据库）
• 人、地点和事物——通过 connectors 将人、家庭或汽车作为节点接入：Telegram（感知消息，发送 send 回复）、Home Assistant （感知整个智能家居的状态变化，通过 call_service 控制设备——灯、门锁、 媒体设备），以及 Smartcar（感知约 30 个品牌的联网车辆——位置/电池/燃油——并执行操作：lock/unlock、充电）

将其指向一个 https://… 端点、postgres://… / mongodb://… / redis://… DSN、grpc://… 目标，或其他 MCP 服务——发现机制会识别接口、了解其结构，并向你的 Agent 提供类型化记录。无论底层是什么，同一套 fetch/query/write 都能正常工作。无需手动为每项服务编写连接器；上游 发生变化时，集成会自动维护。

# 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": "[email protected]", "total_cents": 9900},
                   allow_write=True)               # opt-in; mutates the store

你无需手写连接器或架构：大模型会在设置时学习一次接口（数据库会自行内省，连这一步也省去），上游发生变化时，集成还能自我修复。运行时只是纯粹的确定性传输——成本可预测、行为可复现，无需专人维护。

为 Agent 面临的现实限制而构建

触达所有资源只是一半。另一半是：每个 Token 都要付费，形状不一致会让 Agent 困惑，而且 Agent 无法解析错误说明。Liquid 通过具体的原语逐一解决这些问题——均已发布，均可从 PyPI 获取。

控制上下文预算

# 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

跨来源规范化

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

时间戳（Unix / ISO 8601 / RFC 2822）统一为 UTC datetime；分页封装（{data:[…]} / {results:[…]} / Link 标头）会被展平；ID 字段会在 id / _id / uuid / *_id 之间统一规范。

规范化意图——跨服务采用统一的思维模型

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

结构化恢复——Agent 无需解析文本即可自我修复

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)

每个错误都附带一个 Recovery，其中包含 next_action: ToolCall、retry_safe 和 retry_after_seconds。401 → store_credentials。404/410 → repair_adapter。429 → 按指定的延迟后重试。当上游的架构发生变化时，适配器会自我修复 （repair_adapter）——Agent 继续工作。

成本可预测——调用前就知道成本

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

to_tools() 生成的工具会附带一个 metadata 块（cost_credits、 typical_latency_ms、cached、idempotent、side_effects、related_tools），让 Agent 能够判断该选择哪个工具；而环境工具 （liquid_check_quota、liquid_list_adapters、…）可让它查询状态，无需 死记硬背。

---

实测效果

在真实 Agent 任务上进行的确定性基准测试（500 个订单、200 个工单 测试数据，HTTP 模拟）——可通过 python -m benchmarks.run 重现：

任务	指标	基线	使用 Liquid	变化
查找 10 个金额超过 $100 的订单	tokens	75,482	1,519	−98%
按状态统计收入（聚合）	tokens	75,482	115	−100%
获取客户（仅 id+email）	tokens	424	12	−97%
从 401 错误中恢复	结构化 next_action	否	是	—
查找运输工单	tokens	14,588	154	−99%
Stripe↔PayPal 一致性	字段重叠度	0.11	1.00	+9×
通过估算跳过无效调用	tokens	14,943	0	−100%
max_tokens=2000 预算上限	tokens	14,943	1,999	−87%

完整方法说明及各任务的详细数据：benchmarks/RESULTS.md。

安装

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

是否需要额外配置 LLM？ 自描述接口——OpenAPI、GraphQL、 gRPC、MCP、A2A、WSDL——以及所有数据库（通过内省）均可在无需 LLM 的情况下发现，而且整个运行时（fetch/query/write/sense）都不会调用 模型。只有在发现没有机器可读规范的 REST API（启发式方法 + LLM）以及 映射其字段时，才需要 LLM 后端。[discovery] 会引入 LiteLLM，可连接 OpenAI / Gemini / Anthropic / 本地模型 / 100 多家提供商； 也可以直接选择其中一个：

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.

核心部分不依赖任何外部库——每个后端的库都是可选扩展， 仅在使用时导入。

亲眼看看它如何运行——实时体验，无需预先配置

将 Liquid 指向一个从未接触过的 API（没有适配器、没有 OpenAPI 规范、没有身份验证）， 即可获取类型化记录——你无需编写连接器；只有发现和映射阶段会运行模型。通过 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

你无需编写连接器、架构或身份验证胶水代码——Liquid 会为你学习接口， 并在接口发生变化时重新学习。这正是它的意义：无需你构建或操心维护的集成。

作为 MCP 服务运行（开源、可自托管）

将引擎提供给任意 MCP 客户端（Claude Desktop、Cursor、Claude Code）——它 在你自己的进程中运行，无需云服务、账号，也不会受制于供应商：

在 Cursor 中一键安装（按钮会将服务写入你的 mcp.json；之后在 Cursor 的 MCP 设置中添加 OPENAI_API_KEY）。或者手动设置：

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

使用 uvx 零安装（liquid-mcp 软件包可让你直接按名称运行命令）——Claude Code：

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

Claude Desktop / 任意 MCP 客户端：

{ "mcpServers": { "liquid": {
  "command": "uvx",
  "args": ["liquid-mcp"],
  "env": { "OPENAI_API_KEY": "sk-..." }
} } }

（或者在 pip install liquid-api 后，去掉 uvx，直接使用 "command": "liquid-mcp"。）

**在 Claude Desktop 中一键安装：**安装 .mcpb bundle — 安装时会提示输入模型密钥（存储在操作系统密钥链中），无需编辑 JSON。 机器上需要安装 uv。

工具：liquid_connect（发现并映射任意接口）、liquid_fetch、 liquid_query（服务端搜索/聚合）、liquid_estimate（预先估算 成本/大小，不发起调用）、liquid_list_adapters、liquid_discover。该接口 默认只读；启动服务时设置 LIQUID_ALLOW_WRITES=1，还可 启用 liquid_execute（数据库插入/更新/删除）。适配器和凭据 保存在 ~/.liquid 下。由任意大模型提供支持——OpenAI、Gemini、Anthropic，以及通过 base_url 接入的任意 兼容 OpenAI 的本地端点、通过 LiteLLM 接入的 100 多家提供商， 或通过 CallableBackend 接入你自己的函数。

快速开始——LangGraph Agent

from liquid import Liquid, InMemoryCache, RateLimiter
from liquid._defaults import InMemoryVault, InMemoryAdapterRegistry, CollectorSink
from liquid_langchain import LiquidToolkit
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

liquid = Liquid(
    llm=my_llm, vault=InMemoryVault(), sink=CollectorSink(),
    registry=InMemoryAdapterRegistry(), cache=InMemoryCache(), rate_limiter=RateLimiter(),
    normalize_output=True,    # cross-source canonical shapes
    include_meta=True,        # _meta block on every response
)

adapter = await liquid.get_or_create(
    "https://api.shopify.com",
    target_model={"id": "str", "total_cents": "int", "customer_email": "str"},
    credentials={"access_token": "shpat_..."},
    auto_approve=True,
)

tools = LiquidToolkit(adapter, liquid).get_tools()
agent = create_react_agent(ChatOpenAI(model="gpt-4o-mini"), tools)
result = await agent.ainvoke(
    {"messages": [("user", "Find 5 recent orders over $100 from VIP customers")]}
)

Agent 的工具附带丰富的说明（何时使用、不适用于什么、返回格式、成本）， 每次出错都会提供结构化恢复机制，还支持服务端搜索，因此不会为了找到 5 个订单而 拉取 500 个。

一个 API 适配所有接口

发现机制会识别目标，并为每个端点标记协议；可插拔的传输驱动会运行它——但面向 Agent 的 API（fetch、query、write、映射、恢复、缓存、速率限制）在所有接口中都完全一致。

接口	运行时	写入	安装
REST / HTTP+JSON	✅	✅ 操作（POST/PUT/PATCH/DELETE）	—
GraphQL	✅ 查询 + Relay 分页	✅ mutations	—
SOAP / WSDL	✅ 标准库 XML	—	—
gRPC	✅ unary + server-streaming（reflection）	—	liquid-api[grpc]
WebSocket	✅ 有界批量读取 + 订阅 + 实时 sense	—	liquid-api[ws]
SSE / NDJSON（HTTP 服务端推送）	✅ 有界批量读取 + 实时 sense	—	—
MCP（Agent）	✅ 调用工具 / 读取资源 + 通知 sense	✅ 工具调用	—
A2A（Agent）	✅ 向 AgentCard 技能发送 JSON-RPC message/send	—	—
Postgres（+pgvector）	✅ 表/视图、筛选、分页、向量搜索	✅	liquid-api[pg]
MySQL / MariaDB	✅ 表/视图、筛选、分页	✅	liquid-api[mysql]
SQLite	✅ 表/视图、筛选、分页	✅	—（标准库）
DuckDB	✅ 表/视图、筛选、分页	✅	liquid-api[duckdb]
SQL Server	✅ 表/视图、OFFSET/FETCH 分页	✅	liquid-api[mssql]
Neo4j（图）	✅ 标签/关系类型、属性筛选	✅ 节点 CRUD	liquid-api[neo4j]
MongoDB（文档）	✅ 集合、字段筛选、分页	✅	liquid-api[mongodb]
Redis（键值）	✅ 键空间命名空间、类型化值、SCAN 分页	✅ SET/HSET/DEL	liquid-api[redis]
MQTT（IoT 发布/订阅）	✅ 订阅 → 批量 + 实时 sense	✅ 发布	liquid-api[mqtt]
Modbus（工业）	✅ 寄存器/线圈读取 + 增量轮询 sense	✅ 寄存器/线圈写入	liquid-api[modbus]
OPC UA（工业）	✅ 节点读取 + 原生订阅 sense	✅ 节点写入	liquid-api[opcua]
BACnet（楼宇）	✅ 对象属性读取 + 增量轮询 sense	✅ 属性写入	liquid-api[bacnet]
ADB（Android）	✅ shell 读取 + logcat sense	✅ shell 操作（input/am）	—（系统 adb）
Email — IMAP/SMTP	✅ 按 UID 读取邮箱 + 新邮件 sense	✅ 发送（MIME）	—（标准库）
Email — Gmail API	✅ list/get + history sense	✅ messages.send	—（OAuth2）

读写。 liquid.write(adapter, endpoint, op="insert", values={...}, allow_write=True) 可修改任何数据库（SQL INSERT/UPDATE/DELETE、Mongo insert/update/delete、Redis SET/HSET/DEL、Neo4j 节点 CRUD）；Web/Agent 写入通过经过验证的操作执行。标识符来自内省，值通过参数化处理；update/delete 必须提供 where（不允许不加限制的修改）；写入功能默认关闭，只有显式启用 allow_write=True 后才可用。

Sense — 传入信号的感知器官。 liquid.sense(adapter, endpoint) 可感知任何存在的实时事件流：SQL 行增量（以及 Postgres LISTEN/NOTIFY）、Redis pub/sub、WebSocket 帧、HTTP 服务端推送（SSE/NDJSON）和 MCP 通知 — 每个事件都会作为与模态无关的事件生成。朝向内部时，liquid.sense_webhook(port=…, verifier=…) 会托管一个入站端点，因此服务（或通过 webhook 发送请求的人）向 Agent POST 时，也会成为可感知的信号。所有操作都受 max_events / max_seconds 限制，因此 Agent 可以通过拉取逐步处理事件。

感知运动循环。 react(stream, handler) 会为每个感知到的事件执行处理程序 — 并提供错误隔离和有界并发 — 让宿主可以感知 → 唤醒 Agent → 执行操作。merge_senses(*streams) 会将多个感知流合并到一个循环中，让一个 Agent 可以同时监视数据库、队列和 webhook：

events = merge_senses(
    await liquid.sense(orders, "/orders"),
    await liquid.sense_webhook(port=8088, verifier=stripe_verifier),
)
await react(events, on_event, max_concurrency=4)

自动发现 — 并即时识别。 管道运行前，指纹识别步骤会确定目标：裸格式的 host:port 会根据常见端口规范化（db:5432 → postgresql://db:5432），而 liquid.identify(url) 会回答“这是什么，以及是否已安装其驱动？”；如果缺少后端，还会提供安装提示。（即时识别协议是可行的；但要讲一种新的、经过身份验证的二进制协议则不可行 — 因此，对未知项只会进行命名，不会猜测。）

发现方式	搜索位置	成本
数据库	目录内省（postgres://、mysql://、mongodb://、redis://、neo4j://、…）	低
gRPC / WebSocket / SSE	服务反射 / 帧采样 / 内容类型嗅探	低
MCP / A2A / Plugin	/mcp、/.well-known/agent-card.json、/.well-known/ai-plugin.json	低
OpenAPI / GraphQL / SOAP	规范、内省或 WSDL	低
REST 启发式发现	常见路径 + 大模型解读	中
浏览器	使用 Playwright 捕获网络流量	高

无需编写代码即可添加后端。 对于 SQL 系列，契约足够声明式，可以作为数据处理：通过 register_sql_manifest({...}) 注册方言清单（引号规则、占位符样式、分页、内省 SQL、错误映射、DBAPI2 模块），即可安装可用的驱动和发现功能，因此新的 SQL / 线路兼容存储（CockroachDB、ClickHouse、任何 DBAPI2 驱动）即使是从网络获取的 JSON，也能无需发布新版本即可连接。其他新协议则通过 liquid.transport.ProtocolDriver 协议接入；SQL 后端共用支持方言的核心，因此新增一个后端只需约 80 行适配代码。

想教 Liquid 支持新协议？ 完整的传输驱动（fetch/write/sense）通常约需 150 行代码——请参阅 docs/ADDING_A_DRIVER.md 中的操作指南和待支持协议列表（CAN bus、CoAP、KNX、AMQP、NATS、SNMP、…）。欢迎贡献。

全局目录中已预先发现并映射了 2,500 多个 API——大多数热门服务无需支付发现成本即可连接。

架构

URL / DSN                       Agent
   ↓                              ↑
 FINGERPRINT → DISCOVERY        FETCH · QUERY · WRITE · SEARCH · AGGREGATE
   ↓                              ↑
 one ProtocolDriver per          Deterministic per-protocol transport
 interface:                        • Query DSL (server-side filter)
   REST GraphQL gRPC WS SSE MQTT   • Output normalization
   MCP A2A · SQL graph doc KV ·    • Verbosity / max_tokens / _meta
   Modbus OPC-UA BACnet ADB …      • (full protocol list in the table above)
   ↓                              • Structured recovery + self-heal
 APISchema                        • Rate-limit-aware token bucket
   ↓                              • Response cache (Cache-Control aware)
 AI MAPPING (setup only)          • Empirical probing data (Cloud)
   ↓
 AdapterConfig

AI 仅参与设置阶段。 运行时只使用传输和转换——每次调用都无需使用大模型，成本可预测、行为可复现（search_nl 除外，它会缓存编译结果）。

可替换组件

每个横切关注点都是可替换的 Protocol：

from liquid.protocols import (
    Vault, LLMBackend, DataSink, KnowledgeStore, AdapterRegistry, CacheStore,
)

所有这些组件都提供内存实现；liquid-cloud 为托管部署提供 PostgresVault、RedisCache 等。

框架支持

adapter.to_tools(format="anthropic")   # Claude tool use
adapter.to_tools(format="openai")      # OpenAI function calling (LangChain/CrewAI consume these)
adapter.to_tools(format="mcp")         # MCP (Claude Desktop, Cursor)

框架集成

无需安装额外软件包——此功能已内置于 liquid-api。 adapter.to_tools(format="anthropic" | "openai" | "mcp") 会生成可直接使用的工具定义，适用于 Claude 工具调用、OpenAI 函数调用（LangChain / LangGraph 和 CrewAI 可直接使用），以及任何 MCP 客户端（Claude Desktop、Cursor、…）。随附的 liquid-mcp 服务也开箱即用，可将 Liquid 作为 MCP 工具提供。

对比

功能	Liquid	Zapier	LangChain 工具	DIY
自动发现任何接口（无需精选连接器）	是	否	否	否
在同一层中整合 API、数据库和 Agent	是	部分支持	否	否
通过同一 API 读取并写入	是	是	部分支持	否
服务端搜索 / 聚合	是	否	否	部分支持
跨来源输出规范化	是	部分支持	否	否
提供包含 next_action 的结构化恢复方案	是	否	否	否
模式变更时自动修复	是	否	否	否
预先估算成本	是	否	否	否
原生支持 MCP + A2A + LangChain + CrewAI	是	否	部分支持	否
开源	是	否	是	不适用

文档

• 快速入门 — 发现 → 映射 → 获取，以及无需大模型的运行时
• 开源版与云端版 — 清晰说明免费 / 可自托管与托管服务之间的界限
• 架构
• 扩展 — 实现自己的 Vault / LLM / Sink
• 写入操作规范