thingctx

thingctx turns a description of any API, device, or local process into tools an AI agent can call: authorized per operation, over the system's own transport, and the agent never holds your keys. The description is a W3C Web of Things Thing Description (a TD), a plain JSON file naming what the system can do. Write one, or compile it from an OpenAPI spec. The described system is the server, so you run no server per integration.

A "Thing" is whatever you want to describe: a REST API, an app, a sensor, a media server, a local object, or one system that is several of those at once. A media server takes commands over HTTP and serves its stream over RTSP. The pump in examples/ reads over MQTT and acts through a local call. Each is one Thing and one description.

A description names actions, properties, and events, and gives each one a form: the entry that says where and how to reach it. The URL scheme in that address picks the transport per call, so however many protocols one Thing spans, the agent still sees one tool set.

docs/BINDINGS.md covers the transports and how to add your own. A subprocess transport, exec, also ships, and it ships locked: it refuses every command until you hand it an explicit allowlist.

Those three stay distinct all the way to the call. A read, a write, and a subscribe do not collapse into one undifferentiated function, which is why the gate can allow the read and refuse the write on the same system. A flat list of functions has no such handle.

Pick your path: use it as a library if you own the agent loop, the command line if you do not want to write code, the MCP bridge if your host only speaks MCP, or add a transport if thingctx does not speak your protocol yet.

Stuck, or wondering whether something is possible? Ask in Discussions. Want to build something small? The open issues say what would help.

First run: no keys, no network

One paste shows the whole idea. The description is inline and the handler ships with the package, so nothing here needs a key, a network, or a second file:

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'}

That is the whole loop: a description in, tools out, calls routed. LocalBinding is a binding, a transport implementation; this one routes the local:// address to the handler you pass it. Every other transport works the same way; only the form's href changes.

Install

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

Quote the argument; unquoted brackets fail in zsh. Base pip install thingctx has no dependencies; it already includes the local and exec transports. Redis support is pip install 'thingctx[redis]'. Every optional transport and capability has an extra, listed in pyproject.toml; each one this page uses is named next to the code that needs it.

The document

A description can be small. This one drives the live, key free Open-Meteo forecast API, so it runs as written. Save it as weather.td.json, a later example uses it:

{
  "@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 projects each action to a tool named <thing>__<action>: this file yields weather__forecast (a version segment on the id is dropped). docs/MAPPING.md specifies the full projection. The description carries no secrets, so it is safe to commit and share. The secret is supplied at runtime to the binding that carries the call. Secrets are keyed per Thing.

If your system already has an OpenAPI 3.x spec, you do not have to author a description at all:

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

The spec can be a file or a URL, JSON or YAML; --base-url, --id, and --title override what the spec says. The result is a normal description, so everything else here applies to it unchanged.

Ready made descriptions for common apps, developer tools, and devices live at td.thingctx.com.

Use it as a library

Own the agent loop? Read descriptions, hand the specs to your model, and route each call back through invoke. ThingClient needs no LLM and has no opinion on what chose the action; any code can drive it. This sketch shows the surface against a folder describing a pump; from_arg turns a path or URL into a registry of descriptions:

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
        ...

The form picks the transport per call, so one client can read over HTTP, subscribe over MQTT, and read or write cached properties over Redis. Bindings that pull optional dependencies, including MQTT and Redis, are off by default: install the extra and pass bindings=thingctx.BindingRegistry.default(mqtt=True, redis=True).

Want the loop handled for you? The llm extra runs any provider through litellm; add the http extra, since the weather file's form is HTTPS. Set your provider key in its usual variable (OPENAI_API_KEY, ANTHROPIC_API_KEY, and so on) and pick a model with 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())

The model calls weather__forecast itself and answers in prose:

The forecast for Cairo is clear, around 34 C.

Safety: approval and authorization

Two opt in layers stand between an agent and a real system; both run before any transport is selected. Neither layer handles a credential. The binding holds the secret and the model never sees it.

Approval gates risky calls behind a human. Risk is read from the description, and when to gate is yours: declared (only actions the description marks risky; the default), destructive (adds non idempotent actions and every property write), all, or never. A gated call with no approver is denied: a gate with nobody to open it stays shut. The check sits inside invoke and write_property, so it covers the LLM loop, direct callers, and the MCP bridge alike.

Carrying on from the first example above, with the same 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",
)