thingctx

thingctx

thingctx 可将任意 API、设备或本地进程的描述转换为 AI Agent 可调用的工具：逐项授权，通过系统自身的传输方式调用，并且 Agent 永远不会持有你的密钥。 该描述是 W3C Web of Things Thing Description（简称 TD），一个列出系统功能的普通 JSON 文件。你可以编写一个，也可以从 OpenAPI 规范编译生成。被描述的系统本身就是服务，因此每种集成无需单独运行服务。

“Thing” 是你想描述的任何事物：REST API、应用、传感器、媒体服务、本地对象，或同时具备其中多种特征的系统。媒体服务通过 HTTP 接收命令，并通过 RTSP 提供流。 examples/ 中的水泵通过 MQTT 读取，并通过本地调用执行操作。每个都是一个 Thing，也对应一个描述。

描述会列出 actions、properties 和 events，并为每项提供一个 form：说明如何以及从何处访问该项的入口。该地址中的 URL scheme 决定每次调用所用的传输方式，因此，无论一个 Thing 涉及多少种协议，Agent 看到的始终是一组工具。

docs/BINDINGS.md 介绍了各种传输方式以及如何添加自己的传输方式。项目还提供子进程传输方式 exec，并默认锁定：在你明确提供允许列表之前，它会拒绝所有命令。

这三类操作从始至终都各自独立。读取、写入和订阅不会合并成一个无法区分的函数，因此授权关卡可以允许读取，却拒绝对同一系统的写入。扁平的函数列表无法做到这一点。

选择适合你的路径：将其用作库，适合自行管理 Agent 循环的情况；命令行，适合不想编写代码的情况；MCP 桥接，适合宿主只支持 MCP 的情况；或者添加传输方式，适合 thingctx 尚不支持你的协议的情况。

遇到问题，或想知道某件事是否可行？欢迎在 Discussions 中提问。想做点小贡献？未解决的问题列出了可以帮上忙的事项。

首次运行：无需密钥，无需网络

只需粘贴一次，即可了解整体思路。描述内联编写，处理程序随软件包一同提供，因此这里无需密钥、网络或第二个文件：

pip install thingctx

import asyncio

import thingctx
from thingctx.contrib.time import make_time_handler

TD = {
    "@context": "https://www.w3.org/2022/wot/td/v1.1",
    "id": "urn:thingctx:time",
    "title": "Time",
    "securityDefinitions": {"nosec_sc": {"scheme": "nosec"}},
    "security": ["nosec_sc"],
    "actions": {
        "getCurrentTime": {
            "description": "Current time in an IANA timezone (default UTC).",
            "input": {
                "type": "object",
                "properties": {"timezone": {"type": "string"}},
            },
            "safe": True,
            "idempotent": True,
            "forms": [{"href": "local://getCurrentTime"}],
        }
    },
}


async def main():
    client = thingctx.ThingClient(tds=[TD], bindings=[thingctx.LocalBinding(make_time_handler())])
    tools, invoke = client.as_tools()  # specs for your model; invoke runs a call
    print("tools:", [t["function"]["name"] for t in tools])
    print(await invoke("time__getCurrentTime", {"timezone": "UTC"}))


asyncio.run(main())

tools: ['time__getCurrentTime']
{'timezone': 'UTC', 'datetime': '2026-07-26T15:59:03.233080+00:00', 'utc_offset': '+0000'}

这就是完整的流程：输入描述，生成工具，路由调用。LocalBinding 是一种绑定，也就是传输实现；它会将 local:// 地址路由到你传入的处理程序。其他传输方式的工作原理相同，只有 form 的 href 会变化。

安装

pip install 'thingctx[llm,http,validate]'

请将参数加引号；在 zsh 中不加引号的方括号会导致失败。基础安装 pip install thingctx 不含依赖；它已包含 local 和 exec 传输方式。Redis 支持可通过 pip install 'thingctx[redis]' 安装。每种可选传输方式和功能都对应一个 extra，详见 pyproject.toml；本文用到的每个 extra 都会标注在所需代码旁边。

该文档

描述可以很精简。这个描述驱动的是实时、无需密钥的 Open-Meteo 预报 API，因此可以直接按原样运行。 将它保存为 weather.td.json，后面的示例会用到它：

{
  "@context": "https://www.w3.org/2022/wot/td/v1.1",
  "id": "urn:example:weather:v1",
  "title": "Weather",
  "securityDefinitions": { "nosec_sc": { "scheme": "nosec" } },
  "security": ["nosec_sc"],
  "actions": {
    "forecast": {
      "input": {
        "type": "object",
        "properties": {
          "latitude": { "type": "number" },
          "longitude": { "type": "number" },
          "current": { "type": "string" }
        },
        "required": ["latitude", "longitude"]
      },
      "forms": [{ "href": "https://api.open-meteo.com/v1/forecast", "htv:methodName": "GET" }]
    }
  }
}

thingctx 会将每个操作映射为名为 <thing>__<action> 的工具：此文件会生成 weather__forecast（ID 中的版本段会被去掉）。 docs/MAPPING.md 详细说明了完整的映射规则。该描述不包含任何密钥，因此可以安全地提交和分享。密钥会在运行时提供给承载调用的绑定。密钥按 Thing 单独设置。

如果你的系统已有 OpenAPI 3.x 规范，就无需编写描述：

pip install 'thingctx[openapi]'
thingctx import openapi https://api.example.com/openapi.json --out api.td.json

该规范可以是文件或 URL，格式可以是 JSON 或 YAML；--base-url、--id 和 --title 可覆盖规范中的相应内容。生成的结果是普通描述，因此这里的其他内容对它同样适用。

常见应用、开发工具和设备的现成描述可在 td.thingctx.com 找到。

将其用作库

自己编写 Agent 循环？读取描述，将规范交给模型，再通过 invoke 路由每个调用。ThingClient 无需大模型，也不限制谁来选择操作；任何代码都可以驱动它。下面的示例展示了如何使用一个描述水泵的文件夹；from_arg 可将路径或 URL 转换为描述注册表：

import thingctx


async def run():  # a sketch, not a program: call it from your own loop
    client = thingctx.ThingClient.from_registry(thingctx.from_arg("./descriptions/"))
    specs, invoke = client.as_tools()

    await invoke("pump__set_speed", {"rpm": 1500})
    await client.read_property("pump__rpm")
    await client.write_property("pump__target_rpm", 1500)  # gated like invoke
    async for evt in await client.subscribe(
        "pump__overheat"
    ):  # subscribe returns an async iterator
        ...

表单会为每次调用选择传输方式，因此一个客户端可以通过 HTTP 读取、通过 MQTT 订阅，并通过 Redis 读取或写入缓存属性。需要拉取可选依赖的绑定（包括 MQTT 和 Redis）默认处于关闭状态：安装相应额外依赖，并传入 bindings=thingctx.BindingRegistry.default(mqtt=True, redis=True)。

想让程序替你处理 Agent 循环？llm extra 可通过 litellm 使用任意提供商；由于天气文件中的表单使用 HTTPS，还需添加 http extra。将提供商密钥设置到其惯用的变量中（OPENAI_API_KEY、 ANTHROPIC_API_KEY 等），并使用 THINGCTX_MODEL 选择模型：

import asyncio
import thingctx


async def main():
    # weather.td.json from above; it points at a live, key free API
    host = thingctx.from_file("weather.td.json")
    print(await host.chat("What is the forecast for Cairo? Latitude 30.0, longitude 31.2."))


asyncio.run(main())

模型会自行调用 weather__forecast，并以自然语言作答：

The forecast for Cairo is clear, around 34 C.

安全：审批与授权

在 Agent 与真实系统之间，有两个可选启用的层；它们都会在选择任何传输方式之前运行。这两个层都不处理凭据。密钥由绑定持有，模型永远不会看到它。

审批会要求人工批准有风险的调用。风险取自描述，是否需要审批由你决定：declared（仅审批描述中标记为有风险的操作；默认值）、destructive（增加非幂等操作和所有属性写入）、all 或 never。如果受审批的调用没有审批者，则会被拒绝：无人开启的审批门会一直关闭。检查位于 invoke 和 write_property 内部，因此 LLM 循环、直接调用者和 MCP 桥接都受其保护。

沿用上面的第一个示例，使用相同的 TD：

def approve(req):  # sync or async; return True to allow
    return input(f"run {req.tool_name}{req.arguments}? [y/N] ").lower() == "y"


client = thingctx.ThingClient(
    tds=[TD],
    bindings=[thingctx.LocalBinding(make_time_handler())],
    approve=approve,
    approve_when="all",
)

授权根据策略，按调用者和操作分别作出决定。传入 pdp 和 identity 后，每次调用都会先针对已解析的 (thing, affordance, operation) 进行授权，然后才会到达系统；affordance 是描述中命名的操作、属性或事件。下面的代码片段精简自 examples/14_authz.py，只需安装核心依赖即可直接运行。它定义了下文使用的 TD 和 Pump：一个带有 target_rpm 属性的水泵。

from thingctx import LocalBinding, ThingClient
from thingctx.authz import LocalPolicyGrantSource, PolicyDecisionPoint, build_vocabulary

reader = ThingClient(
    tds=[TD], bindings=[LocalBinding(Pump())]
)  # no pdp yet; parsed only to read the description
vocab = build_vocabulary(reader.things)  # the closed set the description declares

grants = LocalPolicyGrantSource({"operator": {("urn:demo:pump", "target_rpm", "readproperty")}})
pdp = PolicyDecisionPoint(vocabulary=vocab, grant_source=grants)
identity = {"sub": "alice", "roles": ["operator"]}  # claims, validated upstream

client = ThingClient(tds=[TD], bindings=[LocalBinding(Pump())], pdp=pdp, identity=identity)

Alice 的 operator 角色只授予对 target_rpm 的读取权限，因此她的读取操作会通过，写入操作则会被拒绝。可用词汇是封闭的：只有描述中声明的操作才会获得授权。thingctx.authz 不导入加密依赖，并可在无需依赖的核心版本上运行。身份是上游已验证的声明，而不是凭据本身。

运行该示例会输出：

vocabulary (grantable tuples the TD declares):
  (urn:demo:pump, target_rpm, readproperty)
  (urn:demo:pump, target_rpm, writeproperty)

READ  target_rpm  -> ALLOWED, device returned 1200
WRITE target_rpm  -> DENIED, grant does not include ('urn:demo:pump', 'target_rpm', 'writeproperty')

RE-READ target_rpm -> 1200  (unchanged: the denied write never ran)

完整模型见 docs/SECURITY.md。

MCP 桥接器

有些 Agent 只通过 MCP 使用工具。为此，thingctx 提供一个通用 MCP 服务，不为每个集成单独提供服务。它可以提供任何描述注册表：文件夹、URL 或 W3C Thing Description Directory。Directory 是用于描述的标准目录服务。

pip install 'thingctx[mcp,http]'
thingctx-mcp ./descriptions/

对于 Claude Desktop，将以下内容添加到 ~/Library/Application Support/Claude/claude_desktop_config.json （Windows 上为 %APPDATA%\Claude\claude_desktop_config.json），然后重启：

{ "mcpServers": { "things": {
  "command": "thingctx-mcp",
  "args": ["/path/to/your/descriptions/"],
  "env": { "THINGCTX_APPROVE_WHEN": "destructive" } } } }

这里同样适用审批关卡：桥接器会标记有风险的工具，并要求客户端确认。拒绝即会拒绝调用。如果客户端无法显示确认对话框，调用会改为挂起等待处理，桥接器的 approve 工具可在聊天中确认该调用。 THINGCTX_APPROVE_WHEN 用于选择策略，而将 THINGCTX_POLICY 设为 read-only 则会直接拒绝写入和更改状态的操作。

描述中不会包含机密，因此桥接器会从环境变量中读取机密：THINGCTX_TOKEN_<SLUG> 会将机密绑定到具有相应 slug 的 Thing（THINGCTX_TOKEN_GITHUB 用于 github），并按照 Thing 声明的方案应用。

命令行

thingctx 命令可在终端中驱动同一个运行时。将 list 和 invoke 指向任意包含描述的文件夹。将上文首次运行时生成的 TD 保存为 things/time.td.json，然后：

thingctx list ./things/
thingctx invoke ./things/ time__getCurrentTime --arg timezone=UTC

{
  "timezone": "UTC",
  "datetime": "2026-07-26T16:45:57.114817+00:00",
  "utc_offset": "+0000"
}

无需额外配置即可运行，因为时钟的处理程序随软件包提供，并绑定到描述中的 urn:thingctx:time ID。

list 会列出文件夹中各个 Thing 提供的所有工具及其架构； invoke 会运行一个操作。thingctx --help 会列出其他子命令，包括上文展示的 OpenAPI 导入器。invoke 遵循相同的审批关卡。--approve-when 用于选择策略（THINGCTX_APPROVE_WHEN 为回退值），-y 表示无需提示即可批准；如果没有终端且未指定 -y，则会拒绝受审批关卡保护的调用，并返回非零退出码。通过命令行调用的 Agent 获得的保护，与使用该库的 Agent 相同。

记录：审批关卡会记录什么

thingctx 读取传感器、发送命令或拒绝命令时，可以写入相应记录。记录分为三种：

记录	含义	示例
Observation	传感器报告的值	10:04 时温度为 91.0
Actuation	已执行的命令及其结果	将阀门设为 80，执行成功
Refusal	审批关卡不允许执行的命令	Agent X 请求打开阀门 7，遭到拒绝

这些记录中的每个字段都使用已发布词汇表中的术语。SOSA，即 W3C 和 OGC 的传感器本体，定义了 observation 和 actuation 的含义。其他四个词汇表用于补充 SOSA 未涵盖的内容：PROV-O 用于说明请求者，schema.org 用于说明命令如何结束，DCAT 用于说明原始数据的位置，Dublin Core 用于说明审批关卡给出的理由。thingctx 不会自行创造术语。测试会针对发布者提供的文件解析每个术语，如果有术语缺失，测试就会失败。

from thingctx.models.sosa import Observation

obs = Observation(result=91.0, observed_property="temperature")
obs.as_row()  # a flat dict for a table
obs.as_jsonld()  # the same fact, in SOSA terms, for anything that reasons over RDF

只需一个参数即可启用。之后，客户端会将它执行的每次读取、写入、命令、查询和取消，以及审批关卡拒绝的每次操作，都报告为一个 Operation：请求者、操作内容、审批关卡的答复、返回结果和发生时间。记录模型会将这些信息转换为相应记录。SOSA 示例：

from thingctx.models.sosa import recorder

rows: list = []
client = base.guarded(
    pdp, identity={"sub": "urn:agent:planner", "roles": ["operator"]}, record=recorder(rows.append)
)

长时间运行的操作会记录为一个 Actuation：启动时写入尚无结果的记录，结束它的查询或取消操作则会写入带有结果的同一条记录。Operation 位于 thingctx.operations 中，不导入任何内容，并且是运行时与记录模型唯一共享的内容。

记录类是生成的，而非手写的

这三个类来自一个文件： records.yaml，使用 LinkML 编写。该文件列出每个字段、其类型、对应的词汇术语， 以及用于判定两条记录是否相同的字段。运行生成器后，它会写出 Python 类、词汇表和 JSON Schema：

scripts/generate_model.py            # regenerate
scripts/generate_model.py --check    # exit 1 if records.yaml changed and nobody regenerated; CI runs this

目录结构如下。你只需查看三个文件，永远不会接触 _generated：

src/thingctx/models/
    _base.py              hand written. Behaviour every record shares (see below)
    sosa/
        records.yaml      hand written. The model: fields, types, vocabulary terms
        __init__.py       hand written. Exports Observation, Actuation, Refusal
        _generated/       written by the generator. Do not edit
            records.py          the three classes, from gen-pydantic
            records.schema.json JSON Schema, from gen-json-schema
            terms.py            the vocabulary tables, from scripts/generate_model.py

records.py 和 records.schema.json 正是 LinkML 自带命令生成的内容。你可以手动运行 gen-pydantic 或 gen-json-schema，得到完全相同的字节；测试会对此进行检查。terms.py 则由 thingctx 自己的脚本生成， 因为 LinkML 的 RDF 生成器（gen-owl、gen-jsonld-context）会为每个类和字段创建新的术语，而 thingctx 只需要已发布的术语。

_base.py 保存词汇无法描述的行为：记录会拒绝没有时区的时间戳，根据自身内容生成 ID，以便重复写入的相同事实只占一行，并且知道如何将自身转换为表格行或 JSON-LD 文档。它不会提及任何词汇术语。每个生成的类都会继承它。

要为另一种词汇添加模型，请创建 src/thingctx/models/<name>/，在其中放入 records.yaml，运行生成器，并添加一个 recording.py，包含与 SOSA 相同的两个函数：records_for(operation) 和 recorder(write)。生成器、_base.py 和运行时都无需更改。

绑定层永远不会接触词汇

协议绑定通过 HTTP、MQTT 或 rosbridge WebSocket 传输字节。它并不知道有效载荷的含义。上述记录位于绑定层无法导入的层中。pyproject.toml 中的 import-linter 契约会强制执行这一点，因此，导入词汇的绑定会导致构建失败。驱动 Thing 时不会加载记录模型。

扩展

描述一个新系统不需要使用本仓库的任何内容。让 thingctx 学会它尚不了解的功能只需实现一个类，且各部分的工作方式都相同，因此学会一次即可应用于其他部分。

传输是最常见的情况：

class CoapBinding:
    scheme = "coap"

    async def invoke(self, action, form, arguments): ...  # required
    async def read(self, prop, form): ...  # optional

契约由一个 scheme 和一个异步 invoke 组成。如果传输支持，还可以添加 read、write 或 subscribe；运行时会在调用前检查有哪些方法，因此不支持读取的发布订阅传输只需不实现读取方法即可。将绑定传给客户端即可启用。为 thingctx 已提供服务的某个 scheme 注册绑定，会替换内置绑定。

其余部分也采用相同的形式：

要让 thingctx 学会	实现	使用以下方式验证
新传输	ProtocolBinding	assert_binding_contract
新的身份验证方式	CredentialProvider	assert_provider_contract
新的描述来源	Registry	assert_registry_contract
新的媒体引擎	MediaBackend	assert_media_backend_contract
用于为一组设备提供服务的新协议	GatewayBinding	assert_gateway_binding_contract

每个契约都是一个 typing.Protocol：匹配方法即可，无需继承任何内容。@implements(Contract) 会在导入时检查匹配情况，而 thingctx.testing 中的一致性测试套件会检查类型检查器无法验证的运行时行为。examples/13_custom_stack.py 在单个文件中离线实现了每种契约。

你可以将自己编写的内容保留在自己的包中，也可以贡献出来，避免其他人重复编写。仓库外的包通过入口点注册，无需分叉即可被发现，因此只需安装你的包即可：

入口点组	提供	由以下方式发现
thingctx.bindings	一种传输	discover_bindings
thingctx.auth	出站凭据提供程序	discover_auth
thingctx.guards	入站网关防护	discover_guards
thingctx.gateways	用于为一组设备提供服务的总线	discover_gateway_bindings
thingctx.local_handlers	用于处理 local:// Thing 的进程内处理程序	discover_local_handlers

请参阅 docs/BINDINGS.md。

互操作性

thingctx 会使用描述，无论描述由谁生成。 examples/interop/ 下的示例端到端验证了这一点：Eclipse Thingweb 的 node-wot 提供 Thing，thingctx 与之交互；Eclipse Ditto 为数字孪生生成描述，而 thingctx 通过它往返传递状态。Thing Description 是 W3C 推荐标准，因此你在这里编写的文件可移植到任何其他 WoT 使用方，且 thingctx 能读取 W3C Thing Description Directory。

贡献

驱动你自己的系统不需要使用本仓库的任何内容：编写一份描述，然后将 thingctx 指向它。项目真正需要的是它尚不支持的传输，而每种传输只需一个类。

适合入手的地方：

• good first issue 是背景信息齐全、范围明确的任务。留言认领后，任务就会分配给你。
• help wanted 列出了其他所有开放任务。
• CONTRIBUTING.md 说明了绑定的组成方式。

贡献者提交的补丁新增了一条 lint 规则，并弥补了一处测试覆盖缺口。改动小、范围明确，且已合并。

每次提交都需要包含 DCO Signed-off-by 行，可通过 git commit -s 添加。欢迎提交 AI 辅助编写的补丁；每份补丁都会由真人审核并签署。

有问题不一定要提 issue。 Discussions 适合询问“这个已经能做 X 吗”“你们会接受关于 Y 的 PR 吗”，以及如何连接你使用的系统。即使答案是否定的，在那里提问也很有帮助，因为这能让项目了解大家正在尝试做什么。

许可证

Apache 2.0。请参阅 LICENSE。