thingctx

thingctx

thingctxは、あらゆるAPI、デバイス、またはローカルプロセスの記述を、AIエージェントが呼び出せるツールに変換します。操作ごとに認可され、システム独自のトランスポートを介して実行され、エージェントがキーを保持することはありません。 記述にはW3C Web of Things Thing Description（TD）を使用します。システムで実行できることを記した、シンプルなJSONファイルです。自分で作成するか、OpenAPI仕様からコンパイルできます。記述されたシステム自体がサーバーとなるため、連携ごとにサーバーを実行する必要はありません。

「Thing」とは、記述したいものなら何でも指します。REST API、アプリ、センサー、メディアサーバー、ローカルオブジェクト、あるいはこれらのいくつかを兼ねる単一のシステムです。メディアサーバーはHTTPでコマンドを受け取り、RTSPでストリームを提供します。examples/のポンプはMQTTで読み取り、ローカル呼び出しで動作します。どちらも1つのThingであり、1つの記述です。

記述ではactions、properties、eventsを定義し、それぞれにフォームを指定します。フォームとは、どこへ、どのようにアクセスするかを示すエントリです。そのアドレスにあるURLスキームによって呼び出しごとにトランスポートが選ばれるため、1つのThingがいくつのプロトコルにまたがっていても、エージェントに見えるツールセットは1つです。

トランスポートと独自の追加方法については、docs/BINDINGS.mdをご覧ください。サブプロセストランスポートのexecも同梱されており、明示的な許可リストを渡すまで、すべてのコマンドを拒否するようロックされています。

この3つは、呼び出しまで一貫して区別されます。読み取り、書き込み、サブスクライブは、ひとまとめの関数にはなりません。そのため、同じシステムでもゲートによって読み取りを許可し、書き込みを拒否できます。関数を平坦なリストにしただけでは、このような制御はできません。

目的に合わせて選んでください。エージェントのループを自分で管理するならライブラリとして使う、コードを書きたくないならコマンドライン、ホストがMCPのみをサポートするならMCPブリッジ、thingctxがまだ対応していないプロトコルを使うならトランスポートを追加する方法があります。

行き詰まった場合や、何ができるのか知りたい場合は、Discussionsで質問してください。小さなものを作りたい場合は、役に立つ作業が未解決のIssueに記載されています。

初回実行：キーもネットワークも不要

1回貼り付けるだけで、全体の仕組みがわかります。記述はインラインで、ハンドラーはパッケージに同梱されているため、ここではキーもネットワークも別ファイルも必要ありません。

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://アドレスを渡されたハンドラーにルーティングします。他のトランスポートも同じように動作し、変わるのはフォームのhrefだけです。

インストール

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

引数は引用符で囲んでください。引用符で囲まれていない角括弧はzshでエラーになります。基本のpip install thingctxには依存関係がなく、localおよびexecトランスポートがすでに含まれています。Redisをサポートするには、pip install 'thingctx[redis]'を実行します。すべてのオプショントランスポートと機能には、pyproject.tomlに記載された追加パッケージがあります。このページで使用する追加パッケージは、それぞれ必要なコードの横に記載しています。

ドキュメント

説明は小さくてもかまいません。これは、キー不要のライブ 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 にあります。

ライブラリとして使う

エージェントのループを独自に実装しますか？説明を読み込み、仕様をモデルに渡し、各呼び出しを invoke 経由でルーティングしてください。ThingClient は LLM を必要とせず、何がアクションを選んだかにも関与しません。どのようなコードからでも操作できます。この例では、ポンプを説明するフォルダーを使った基本的な使い方を示します。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
        ...

フォームでは呼び出しごとにトランスポートを選べるため、1 つのクライアントで HTTP 経由の読み取り、MQTT 経由のサブスクライブ、Redis 経由のキャッシュされたプロパティの読み取りや書き込みを行えます。MQTT や Redis など、オプションの依存関係を読み込むバインディングは既定で無効です。追加機能をインストールし、bindings=thingctx.BindingRegistry.default(mqtt=True, redis=True) を渡してください。

ループを自動で処理したい場合は、llm extra を使うと litellm 経由で任意のプロバイダーを実行できます。weather ファイルのフォームは 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.

安全性：承認と認可

エージェントと実システムの間には、オプトインのレイヤーが 2 つあります。どちらもトランスポートが選択される前に実行されます。どちらのレイヤーも認証情報を扱いません。シークレットはバインディングが保持し、モデルには見えません。

承認では、リスクの高い呼び出しを人間の確認に回します。リスクは説明から読み取られ、確認を必要とする条件は利用者が選べます。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 というプロパティが 1 つあります。

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 は暗号関連のパッケージをインポートせず、依存関係のないコア環境で実行できます。identity は上流で検証された情報であり、認証情報そのものではありません。

例を実行すると、次のように出力されます。

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 ブリッジ

MCP 経由でツールを受け取るエージェントもあります。そのようなエージェント向けに、thingctx は汎用 MCP サーバーを 1 つ提供しており、インテグレーションごとのサーバーはありません。フォルダー、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> は、スラッグが一致する 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 は 1 つのアクションを実行します。thingctx --help には、上記の OpenAPI インポーターを含むその他のサブコマンドが表示されます。invoke にも同じ承認ゲートが適用されます。--approve-when でポリシーを選択し（フォールバックは THINGCTX_APPROVE_WHEN）、-y でプロンプトなしに承認します。ターミナルがなく、-y も指定されていない場合、承認ゲートの対象となる呼び出しは、ゼロ以外の終了コードで拒否されます。エージェントがシェルからコマンドを実行する場合も、ライブラリを使う場合と同じ保護が適用されます。

記録：ゲートが書き残す内容

thingctx は、センサーを読み取るとき、コマンドを送るとき、またはコマンドを拒否するときに、その内容を記録できます。記録には次の 3 種類があります。

記録	意味	例
Observation	センサーが報告した値	10:04 の温度は 91.0 でした
Actuation	実行されたコマンドとその結果	バルブを 80 に設定し、成功しました
Refusal	ゲートが許可しなかったコマンド	エージェント X がバルブ 7 を開くよう求めましたが、拒否されました

これらの記録のすべてのフィールドには、公開された語彙の用語が使われます。センサーのオントロジーである W3C と OGC の SOSA は、観測と作動の定義を定めています。SOSA に不足している部分は、ほかの 4 つの語彙で補います。誰が要求したかは 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

引数を 1 つ指定すると有効になります。その後、クライアントは実行するすべての読み取り、書き込み、コマンド、クエリ、キャンセルと、ゲートが拒否したすべての操作を Operation として報告します。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)
)

長時間実行されるアクションは 1 つの Actuation です。開始時に結果を含まない記録を書き込み、そのアクションを終了させるクエリまたはキャンセルの際に、同じ記録に結果を書き込みます。Operation は thingctx.operations にあり、何もインポートせず、ランタイムと記録モデルが共有する唯一のものです。

レコードクラスは手書きではなく生成されます

3つのクラスは、LinkMLで書かれた1つのファイル、 records.yamlから生成されます。このファイルには各フィールド、その型、対応する語彙用語、そして2つのレコードが同じレコードであると判定するためのフィールドが記載されています。ジェネレーターを実行すると、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

ディレクトリは次のとおりです。読み込むファイルは3つで、_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を導出することで、同じ事実を2回書き込んでも1行になり、テーブルの行やJSON-LDドキュメントへの変換方法も把握しています。語彙用語には一切言及しません。生成された各クラスは、このクラスを継承します。

別の語彙用のモデルを追加するには、src/thingctx/models/<name>/を作成し、その中にrecords.yamlを配置してジェネレーターを実行し、SOSAと同じ2つの関数records_for(operation)とrecorder(write)を含むrecording.pyを追加します。ジェネレーター、_base.py、ランタイムを変更する必要はありません。

バインディングは語彙を認識しません

プロトコルバインディングは、HTTP、MQTT、またはrosbridge WebSocket経由でバイト列をやり取りします。ペイロードの意味は把握しません。上記のレコードは、バインディングがインポートできないレイヤーに置かれています。pyproject.tomlのimport-linterコントラクトによってこれが強制されているため、語彙をインポートするバインディングがあるとビルドに失敗します。Thingを操作するときにレコードモデルが読み込まれることはありません。

拡張する

新しいシステムの説明には、このリポジトリの何も必要ありません。thingctx に まだ知らないことを教える方法は 1 つのクラスで、どの部分も同じように 動作するため、一度学べば、ほかの部分にも応用できます。

トランスポートの場合は、次のようになります。

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 がすでに提供しているスキームにバインディングを登録すると、組み込みの バインディングが置き換えられます。

ほかの部分も同じ形式です。

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 では、オフラインで 1 つのファイル内に各実装を作成しています。

作成したものは自分のパッケージ内で非公開にすることも、同じものが二度作られないよう コントリビュートすることもできます。外部パッケージはエントリーポイント経由で登録され、 フォークなしで検出されるため、インストールするだけで使えます。

エントリー・ポイント・グループ	提供するもの	検出関数
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 に渡すだけです。プロジェクトが必要としているのは、まだ対応できていない トランスポートです。それぞれ 1 つのクラスで実装できます。

着手しやすい項目は次のとおりです。

• good first issue は、背景情報が記載された範囲の明確な作業です。担当したい項目にコメントすれば、 あなたに割り当てられます。
• help wanted には、 そのほかの対応可能な項目があります。
• CONTRIBUTING.md では、バインディングの仕組みを説明しています。

コントリビューターのパッチによって、リンティングルールが追加され、カバレッジの 不足が解消されました。小さく範囲を絞った変更がマージされています。

すべてのコミットには DCO の Signed-off-by 行が必要です。git commit -s で追加できます。 AI の支援を受けたパッチも歓迎します。実際の人間がレビューして署名します。

質問に issue は必要ありません。 Discussions は、「これはすでに X に 対応していますか」「Y の PR を受け入れてもらえますか」、または手元にあるシステムへの 接続方法について話す場所です。回答が「いいえ」の場合でも、質問することには意味があります。 プロジェクトに、利用者が何をしようとしているかが伝わるからです。

ライセンス

Apache 2.0。LICENSE をご覧ください。