--- name: toolchestrator-connect description: Connect the current tool/project to a Toolchestrator hub. Use when the user says "connect this tool to Toolchestrator", "register this with the hub", "add Toolchestrator to this project", "publish this tool's data to the hub", "make this tool orchestrable", "let others use this tool", "expose this tool's UI through the hub", or similar. Covers installing the toolchestrator Python client from the hub, registering the tool, syncing versioned data-model schemas and records, reading other tools' data, serving remote task actions, invoking other tools synchronously, and exposing a tool's local web UI through the hub. --- # Connect a tool to Toolchestrator Toolchestrator is a company hub for the small tools people build with Claude Code. Tools stay on their authors' machines; the hub is the durable layer. A connected tool: 1. appears in the company **registry** (name, author, problem it solves), 2. publishes **versioned JSON Schemas** for its data models, 3. optionally **syncs records** to the hub, scoped to owner / department / company, 4. can **execute tasks** queued by people or other tools (outbound long-poll `serve` — the hub never connects into this machine), 5. can **read other tools' data** through the hub with filtered queries; every real read appears as a **connection** on the company data map, 6. can **invoke another tool's action and get the result back** (`tc.call`), 7. can **expose its own local web UI** so other users drive it through the hub (`tc.serve_web` — still outbound-only from this machine). You integrate all of this with the pip-installable `toolchestrator` client (dependencies: `requests` + `websocket-client`; works on Python 3.9+). ## What leaves this machine (data handling) Be explicit with the user about what connecting sends to the hub, and confirm they're comfortable before syncing anything beyond the tool's description: - **What the hub receives and stores at rest:** the tool's description / author / problem text; the JSON Schemas of the models you sync; the records of those models (`sync_data`); the payloads and results of tasks the tool runs; and an activity log of these events. - **What only passes through:** web-UI traffic tunnelled by `serve_web` is relayed through the hub to reach teammates' browsers; the hub does not retain it beyond delivering it. - **What never leaves:** the hub never connects into this machine, and it only ever receives what the tool pushes — it cannot read files, env, or anything you don't sync. **Never sync secrets, credentials, API keys, or tokens** (strip them before `sync_data`, even at private scope — the hub is a collaboration layer, not a secret store). - **Who can see it:** the tool's sharing scope (`private` / `department` / `company`) governs who can read the data, run the actions, and open the app, and the hub enforces it **server-side on every request** — not just in the UI. Start `private`; widen deliberately (see Guardrails). The hub's `/privacy` (or the operator) has retention/deletion details. ## The hub URL always comes from the user — never discover it **Do not search for, scan for, probe, ping, or assume a local Toolchestrator instance.** The hub is a remote SaaS the user's company runs (e.g. `https://hub.toolchestrator.com`). It is almost never on this machine, so: - **Never** `curl`/connect to `localhost`, `127.0.0.1`, or `:8787` to look for a hub, and never default the URL to localhost. - **Never** guess the address from other files, git remotes, or the environment. - **Always ask the user** for the hub URL (and the personal token below) and use exactly what they give you. If they haven't provided it, stop and ask — do not proceed on a guess. ## Step 1 — Install the client The `toolchestrator` client is published on PyPI, so installing it needs nothing but a package manager: ```bash uv pip install toolchestrator # or: pip install toolchestrator ``` It works on Python 3.9+ and pulls its only dependencies (`requests`, `websocket-client`) automatically. Install it into whatever environment the tool runs in. You also need **two things from the user** for registration (Step 3) — ask now if you don't have them: 1. **Hub URL** — the company's Toolchestrator address, e.g. `https://hub.toolchestrator.com` (see the rule above — never assume one). 2. **Personal access token** — starts with `tcu_`, created in the dashboard under **Settings → Personal tokens**. Needed once, for registration. **Never ask for the user's password**; the token exists so you never handle one. **Offline, air-gapped, or white-label hub that can't reach PyPI?** The hub serves the same client itself. Fetch the fallback command from the hub (unauthenticated) and run it: ```bash curl -s /api/client # -> run the "install_from_hub" command it returns ``` ## Step 2 — Understand the tool before registering it Read the project's README and main modules, then draft — and confirm with the user — three short texts. They are the tool's face in the company catalog, so make them concrete: - `description`: what the tool does, one or two sentences. - `problem_statement`: the pain it removes and for whom. - `usage`: how a colleague (or another tool) would actually use it. Also decide which **actions** the tool should execute remotely (e.g. `mark_paid`, `rescore`). If you wire `serve` in Step 6, the action list stays in sync from the handler keys automatically. ## Step 3 — Register (one time) Run this once (a tiny script, or a `register` subcommand you add to the tool). Run it from the **project root** — it writes `.toolchestrator.json` in the current directory. Use the hub URL the user gave you (shown here as a placeholder — do not hardcode localhost): ```python from toolchestrator import Toolchestrator tc = Toolchestrator.register( "https://your-hub.example.com", # base_url — the address the user gave you "tcu_...", # personal_token (from the user) "invoice-radar", # slug: url-friendly, unique per company "Invoice Radar", # name description="Tracks vendor invoices and flags overdue ones.", problem_statement="Finance had no shared view of unpaid invoices.", usage="Sync runs after each import; enqueue mark_paid with {'invoice_id': ...}.", sharing_scope="private", # start private — see Guardrails department=None, # or a department name like "Finance" actions=["mark_paid"], # optional; serve() maintains this later ) ``` - The hub returns a per-tool API key (`tck_...`); `register` saves it with the hub URL, tool id, and slug into `.toolchestrator.json`. - If the slug is already registered by this user (e.g. re-running on a new machine), pass `reconnect=True` to rotate the key and update metadata; otherwise the hub answers 409. - **Immediately add `.toolchestrator.json` to `.gitignore`** — it holds the API key. Do it in the same edit, not as a follow-up. Afterwards, everywhere else in the tool just do: ```python tc = Toolchestrator() # finds .toolchestrator.json walking up from cwd; # env TOOLCHESTRATOR_URL / TOOLCHESTRATOR_API_KEY override ``` ## Step 4 — Survey the ecosystem (always) **Before you design or change this tool's data model, look at what the company already has.** Do this every time — a brand-new tool and an edit to an existing one both start here. The whole point of the hub is one shared model per concept: a second tool that re-invents `invoice` or `customer` fragments the data map instead of joining it. - **Discover what exists.** List the visible tools and their models, then read the actual fields of anything close to what you were about to build (`schemas` with no argument returns this tool's own models): ```python tc.list_tools() # visible tools + their model names tc.schemas("invoice-radar") # that tool's models -> versions -> json_schema (its fields) tc.schemas() # this tool's own models (tool omitted = self) ``` - **Reuse over rebuild.** If another tool already owns the concept, do **not** publish a duplicate model — consume theirs: declare the connection and read it (Step 7). Publish a new model only for a concept no existing tool owns. When two tools need overlapping data, the one that produces it publishes; the other reads. ```python tc.connect_to("invoice-radar", "invoice", purpose="...") # then tc.read(...) — see Step 7 ``` - **Editing a tool that already publishes? Learn who depends on it first.** Before you touch a model this tool owns, see which other tools read it and exactly which fields they take: ```python tc.dependents() # -> {"models": [{"model": "invoice", "latest_version": 3, # "dependents": [{"tool": {"slug": "todo-tool", "name": "Todo Tool"}, # "fields": ["id", "vendor", "amount"], "all_fields": False, # "purpose": "...", "reads_count": 12, ...}]}]} ``` Those fields are **contracts** — any rename/removal/retype of one breaks that consumer. See the **Soft-locked fields** guardrail before changing them. ## Step 5 — Sync schemas and data Publish a schema only for the models this tool genuinely owns (you checked Step 4 for one to reuse first). Find where the tool stores its records (SQLite table, JSON file, dataframe...) and derive a JSON Schema per model. Push the schema at startup or wherever the storage shape is defined — the hub versions it, identical pushes are no-ops, and that is how schema drift is tracked: ```python INVOICE_SCHEMA = { "type": "object", "properties": { "id": {"type": "integer"}, "vendor": {"type": "string"}, "amount": {"type": "number"}, "status": {"type": "string", "enum": ["open", "paid", "overdue"]}, }, "required": ["id", "vendor", "amount"], } tc.sync_schema("invoice", INVOICE_SCHEMA) # -> {"model": "invoice", "version": 1, "changed": True, "warnings": [...]} ``` When a push creates a new version, `sync_schema` also returns `warnings` — the hub names any field it changed or removed that another tool depends on, and the affected consumer's slug. The version is never blocked, but a warning means you are about to break someone: **never ignore it** (see the **Soft-locked fields** guardrail). Adding fields is always safe and warns nobody. Then sync records at the tool's natural "data changed" points. `sync_data` upserts by a stable id, so syncing the full current state repeatedly is fine: ```python tc.sync_data("invoice", invoices, id_field="id") # invoices: list of dicts tc.sync_data("invoice", changed_rows, deleted_ids=["42"]) # incremental variant ``` `sync_schema` must run before the first `sync_data` for a model. Sync only the models worth sharing — not internal caches, and **never** secrets (see Guardrails). ## Step 6 — Serve actions (make the tool orchestrable) Wire a `serve` subcommand so the tool can execute tasks queued on the hub — by people (the dashboard **Run** button / Tasks tab) or by other tools (`tc.call`, `tc.enqueue`): ```python def mark_paid(payload, task): """payload: the caller's dict; task: full task dict (id, attempt, ...).""" inv = db.mark_paid(payload["invoice_id"]) tc.sync_data("invoice", [inv], id_field="id") # keep the hub fresh return {"invoice_id": inv["id"], "status": inv["status"]} tc.serve({"mark_paid": mark_paid}) # blocking; Ctrl-C to stop ``` Handler contract: - return a dict (or `None`) → task **succeeded**, return value is the result; - raise → task **failed** with the exception message (the hub retries while attempts remain); - an action with no handler → task fails with `no handler for `. `serve` first PATCHes the tool's `actions` to the handler keys, so the dashboard's Run form always matches reality. Polling also keeps the tool marked **online**. For tools that shouldn't block (cron-style), use `tc.serve_once(handlers, wait=5)` — one poll, returns the number of tasks handled. ## Step 7 — Optional: consume and drive other tools Discover what's visible, then declare → read narrowly → react. Everything here obeys the same visibility scopes — the tool acts as its owner, so it sees what its owner may see. ```python tc.list_tools(q="invoice") # discover visible tools + their models tc.schemas("invoice-radar") # inspect that tool's fields before you read/reuse ``` **Declare the connection before reading** (idempotent; do it at startup next to `sync_schema`). Reads are auto-observed either way, but a declared connection with a purpose is what makes the company data map explain *why* this tool reads that data: ```python tc.connect_to("invoice-radar", "invoice", purpose="Create follow-up todos for overdue invoices") tc.connections() # this tool's connections + read stats ``` **Read with `where`/`fields`** instead of dumping whole models — the hub filters, projects, and orders: ```python rows = tc.read("invoice-radar", "invoice", where={"status": "overdue", "amount": {"$gte": 1000}}, fields=["id", "vendor", "amount"], order_by="amount:desc", limit=50) # -> [{"id": ..., "vendor": ..., "amount": ..., "_external_id": ..., "_updated_at": ...}, ...] ``` `where` keys are top-level payload fields, AND-ed; a scalar means equality, a dict uses operators `$eq $ne $gt $gte $lt $lte $in $contains` (`$contains` = case-insensitive substring). `order_by` is `""`, `":asc"`, or `":desc"`. (`tc.fetch_data` still works and delegates to `read`; prefer `read` with filters for anything new.) **Invoke another tool's action.** Two ways: ```python # Synchronous: enqueue, wait, and get the result back (RPC through the hub). result = tc.call("invoice-radar", "mark_paid", {"invoice_id": 42}, wait=30) # -> the target's result dict; raises ToolchestratorError if the app fails, or if it did # not finish in `wait` seconds / the target is offline (the exception carries .task_id # so you can poll get_task later). # Fire-and-forget: enqueue and check later. task = tc.enqueue("invoice-radar", "mark_paid", {"invoice_id": 42}) tc.get_task(task["id"]) # status / result ``` Use `tc.call` when you need the answer now (the target must be running its `serve` loop); use `tc.enqueue` for eventual execution. **React to changes with `data_handlers`.** Subscribe, then give `serve` a per-(source, model) handler: ```python tc.subscribe(source_tool="invoice-radar", model="invoice") def on_invoices_changed(payload, task): # payload: {"source_tool": , "model_name", "upserted", "deleted"} refresh_from_invoices() return {"refreshed": True} tc.serve(handlers={"mark_paid": mark_paid}, data_handlers={("invoice-radar", "invoice"): on_invoices_changed}) ``` `data.changed` tasks route by `(payload["source_tool"], payload["model_name"])`; one that matches no `data_handlers` entry falls back to `handlers["data.changed"]` if present, else fails as unhandled. ## Step 8 — Optional: expose the tool's web UI through the hub If the tool has (or gains) a **local web UI**, other users can use it through the hub — the app runs on this machine and their browser reaches it via the hub, with no inbound connection here. Run the tool's web server locally, then open the tunnel alongside `serve`: ```python tc.serve_web("http://127.0.0.1:5000") # blocking; forwards hub → this local web app ``` - `serve_web` opens one **outbound WebSocket** to the hub and forwards requests to the local target you name — it only ever hits that fixed target (never a host from the request). - Run it in a separate thread/process from `serve` (one handles web traffic, the other handles task actions). Make the web app's asset/link/form URLs **relative** so they resolve under the hub's app path. - In the dashboard, a user who can see the tool clicks **Open app** to use it. Who may open it is governed by the tool's sharing scope, exactly like data and actions. - The tunnel is a WebSocket, so `serve_web` needs `websocket-client` (installed automatically by the hub-served install in Step 1). ## Step 9 — Verify 1. `tc.heartbeat()` returns `{"ok": True}`. 2. The tool appears in the dashboard (Tools page) with your texts and models. 3. Enqueue a test task from the dashboard (Run) while `serve` runs; watch it succeed. 4. If you wired `serve_web`, click **Open app** and confirm the UI loads. ## Worked minimal example A complete connected tool in one file — publishes its own `todo` model, and (having surveyed the ecosystem, Step 4) **reuses** `invoice-radar`'s existing `invoice` model instead of duplicating it: it declares the connection, reads it filtered, and reacts via `data_handlers`, rather than publishing its own `invoice`. It also serves an action. The hub URL is passed in, never assumed: ```python #!/usr/bin/env python3 """todo_tool.py — a tiny todo list connected to Toolchestrator.""" import json, sys from toolchestrator import Toolchestrator, ToolchestratorError DB = "todos.json" SCHEMA = {"type": "object", "properties": {"id": {"type": "integer"}, "title": {"type": "string"}, "done": {"type": "boolean"}}, "required": ["id", "title"]} def load(): try: with open(DB) as f: return json.load(f) except FileNotFoundError: return [] def save(todos): with open(DB, "w") as f: json.dump(todos, f) def register(hub_url, token): # hub_url + token come from the user, never guessed Toolchestrator.register( hub_url, token, "todo-tool", "Todo Tool", description="A tiny shared todo list.", problem_statement="Todos lived only in one person's terminal.", usage="Enqueue complete_todo with {'id': }; browse the todo model.", sharing_scope="private") print("registered — now add .toolchestrator.json to .gitignore") def sync(tc): tc.sync_schema("todo", SCHEMA) tc.sync_data("todo", load(), id_field="id") def refresh_overdue_todos(tc): """Read only what we need from invoice-radar; one todo per overdue invoice.""" overdue = tc.read("invoice-radar", "invoice", where={"status": "overdue"}, fields=["id", "vendor", "amount"], order_by="amount:desc", limit=50) todos = load() next_id = max((t["id"] for t in todos), default=0) + 1 for inv in overdue: title = "Chase %s invoice %s" % (inv["vendor"], inv["id"]) if not any(t["title"] == title for t in todos): todos.append({"id": next_id, "title": title, "done": False}) next_id += 1 save(todos) tc.sync_data("todo", todos, id_field="id") return {"overdue_seen": len(overdue)} def complete_todo(payload, task): todos = load() for t in todos: if t["id"] == payload["id"]: t["done"] = True save(todos) Toolchestrator().sync_data("todo", [t], id_field="id") return {"id": t["id"], "done": True} raise ValueError("no todo with id %s" % payload["id"]) def on_invoices_changed(payload, task): return refresh_overdue_todos(Toolchestrator()) if __name__ == "__main__": cmd = sys.argv[1] if len(sys.argv) > 1 else "sync" if cmd == "register": register(sys.argv[2], sys.argv[3]) # todo_tool.py register elif cmd == "sync": sync(Toolchestrator()) elif cmd == "serve": tc = Toolchestrator() sync(tc) try: # declare + subscribe are idempotent; fine to run every start # reuse invoice-radar's model (found via list_tools/schemas) — don't re-publish invoice tc.connect_to("invoice-radar", "invoice", purpose="Create follow-up todos for overdue invoices") refresh_overdue_todos(tc) tc.subscribe(source_tool="invoice-radar", model="invoice") except ToolchestratorError: pass # source not visible / already subscribed — todos still work tc.serve({"complete_todo": complete_todo}, data_handlers={("invoice-radar", "invoice"): on_invoices_changed}) ``` ## Guardrails **The hub URL is user-supplied.** Never scan/probe/assume a local hub, never default to localhost, never infer the address from the environment — ask the user and use their answer. **Soft-locked fields — a field another tool reads is a contract.** - **Adding** fields to a model is always safe — do it freely, it breaks and warns nobody. - **Renaming, removing, or changing the type/enum** of a field a consumer reads **breaks** that consumer. Before such a change, run `tc.dependents()` to see who reads that field (an `all_fields` consumer depends on *every* current field of the model). If any tool reads it, **STOP**: tell the user exactly which tool(s) break and how, and proceed only with their explicit confirmation. - `tc.sync_schema` returns `warnings` naming each breaking field change and the affected consumer, and the client logs them. **Never ignore a sync_schema warning** — surface it to the user the same way; if it was unexpected, roll the change back. **Sharing scope — private first.** - Default to `sharing_scope="private"`. Only the owner (and the tool itself) sees a private tool, its data, its actions, and its web UI. - Escalate to `department` or `company` only when the user explicitly wants colleagues to use the tool — and say what widening means: `department` = everyone in that department can read the synced records, queue/invoke actions, and open the web UI; `company` = everyone in the company. Scope is the single control over who can *use* the tool, including via `Open app`. - **PII check before `company`**: if the synced models contain personal data (names, emails, salaries, addresses, health data, customer records), warn the user explicitly before choosing `company` scope and get their confirmation. When in doubt, stay narrower or strip the sensitive fields. **Never sync secrets.** - Never include API keys, passwords, tokens, credentials, private keys, or password hashes in synced records or schemas — strip those fields before `sync_data`, even at private scope. The hub is not a secret store. **Protect the tool's credentials.** - `.toolchestrator.json` holds the tool's API key: add it to `.gitignore` in the same change that creates it, and never print, log, or commit the `tck_...` / `tcu_...` values. - The personal token (`tcu_`) is used once, for `register`; do not persist it in code, config, or shell history you write. **Scope of change.** - Integration is additive: register/sync/serve/serve_web hooks. Don't refactor the tool's core logic to fit Toolchestrator, and don't make network calls in hot paths that previously worked offline — sync at natural boundaries and keep the tool fully functional when the hub is unreachable (catch `ToolchestratorError` around non-essential syncs).