From c0b2b1f03b62cdd6e88b7660e55c379aeba5142f Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Tue, 30 Jun 2026 21:54:15 -0400 Subject: [PATCH 1/4] adding mcphub and tool universe Signed-off-by: Dhaval Patel --- docs/tool_universe.md | 51 +++++ examples/quickstart_tooluniverse.py | 43 +++++ pyproject.toml | 2 +- src/mcphub/__init__.py | 285 ++++++++++++++++++++++++++++ src/mcphub/workflows.py | 66 +++++++ 5 files changed, 446 insertions(+), 1 deletion(-) create mode 100644 docs/tool_universe.md create mode 100644 examples/quickstart_tooluniverse.py create mode 100644 src/mcphub/__init__.py create mode 100644 src/mcphub/workflows.py diff --git a/docs/tool_universe.md b/docs/tool_universe.md new file mode 100644 index 00000000..e0425607 --- /dev/null +++ b/docs/tool_universe.md @@ -0,0 +1,51 @@ +# Three-step contract (same as ToolUniverse) + +```python +from mcphub import ToolUniverse + +tu = ToolUniverse() # 1. init +tu.load_tools() # 2. load (connect + discover) +tu.run({ # 3. run + "name": "iot.sensors", + "arguments": {"site_name": "MAIN", "asset_id": "Chiller 6"}, +}) +tu.close() +``` + +`load_tools(servers=[...])` limits to specific servers. Tools are namespaced +`.`; a bare name (e.g. `sensors`) also works when unambiguous. +A shorthand `tu.run("iot.sensors", {...})` is accepted too. + +## Discovery + +```python +tu.find_tools("failure mode") # keyword search over loaded tools +tu.list_tools() # all loaded tool + workflow names +tu.list_tools("fmsr") # tool names for one server +tu.tool_specification("iot.sensors") +``` + +## Workflows + +Composed workflows run through the **same `run` entrypoint**: + +```python +tu.run({"name": "chiller_triage", "arguments": {"asset_id": "Chiller 6"}}) +``` + +Add one by writing `fn(tu, **arguments)` in `workflows.py` and listing its name +in `REGISTERED` (or `tu.register_workflow("name", fn)` at runtime). Built in: +`chiller_triage` (sensors → failure modes → mapping → work order) and +`sensor_inventory_gap` (installed vs measured sensors). + +## Run + +```bash +uv sync +docker compose -f src/couchdb/docker-compose.yaml up -d # iot / wo data +cp .env.public .env # WATSONX_* for fmsr +uv run python examples/quickstart_tooluniverse.py +``` + +Each server launches via `uv run -mcp-server` and inherits your +environment. Override with `ToolUniverse(servers={...})` if you run outside `uv`. \ No newline at end of file diff --git a/examples/quickstart_tooluniverse.py b/examples/quickstart_tooluniverse.py new file mode 100644 index 00000000..9d20dc31 --- /dev/null +++ b/examples/quickstart_tooluniverse.py @@ -0,0 +1,43 @@ +""" +Quick start — ToolUniverse-style interface. Prereqs (repo root): + uv sync + docker compose -f src/couchdb/docker-compose.yaml up -d + cp .env.public .env # WATSONX_* for fmsr +Run: + uv run python examples/quickstart.py +""" +import json + +from mcphub import ToolUniverse + + +def show(t, o): + print(f"\n=== {t} ===") + print(json.dumps(o, indent=2, default=str)) + + +def main(): + tu = ToolUniverse() # 1. init + try: + tu.load_tools(servers=["iot", "fmsr", "wo"]) # 2. load (connect + discover) + + show("find 'failure mode'", + [s["name"] for s in tu.find_tools("failure mode")]) + + # 3. run a single tool (ToolUniverse dict form) + show("iot.sensors", tu.run({ + "name": "iot.sensors", + "arguments": {"site_name": "MAIN", "asset_id": "Chiller 6"}, + })) + + # A workflow runs through the same entrypoint + show("chiller_triage", tu.run({ + "name": "chiller_triage", + "arguments": {"asset_id": "Chiller 6", "raise_work_order": False}, + })) + finally: + tu.close() + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 196d0d09..93de1644 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -3,7 +3,7 @@ requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["src/agent", "src/evaluation", "src/llm", "src/observability", "src/servers"] +packages = ["src/agent", "src/evaluation", "src/llm", "src/observability", "src/servers", "src/mcphub"] [project] name = "assetopsbench-mcp" diff --git a/src/mcphub/__init__.py b/src/mcphub/__init__.py new file mode 100644 index 00000000..5250a01f --- /dev/null +++ b/src/mcphub/__init__.py @@ -0,0 +1,285 @@ +""" +mcphub — a ToolUniverse-style client over the AssetOpsBench MCP servers. + +Same interface as ToolUniverse (load_tools -> run), implemented directly over +the six stdio FastMCP servers (iot, utilities, fmsr, wo, tsfm, vibration). +Depends only on the ``mcp`` client the project already requires — there is no +dependency on any external ``tooluniverse`` package. + + from mcphub import ToolUniverse + + tu = ToolUniverse() # 1. init + tu.load_tools() # 2. load (connect + discover) + tu.run({ # 3. run + "name": "iot.sensors", + "arguments": {"site_name": "MAIN", "asset_id": "Chiller 6"}, + }) + +Tools are namespaced ``.``; a bare tool name also works when it is +unambiguous across loaded servers. Workflows are registered and callable through +the same ``run`` entrypoint (see ``mcphub.workflows``). +""" +from __future__ import annotations + +import asyncio +import concurrent.futures +import json +import logging +import os +import threading +from contextlib import AsyncExitStack +from typing import Any, Callable, Dict, List, Optional + +from mcp import ClientSession, StdioServerParameters +from mcp.client.stdio import stdio_client + +__all__ = ["ToolUniverse", "ToolClient", "DEFAULT_SERVERS"] + +_log = logging.getLogger("mcphub") + +# How to launch each server as a stdio process. `uv run` resolves the project +# venv + .env; pass your own table if you run inside an already-active venv. +_SERVER_NAMES = ["iot", "utilities", "fmsr", "wo", "tsfm", "vibration"] +DEFAULT_SERVERS = {n: ["uv", "run", f"{n}-mcp-server"] for n in _SERVER_NAMES} + + +def _unwrap(result) -> Any: + """MCP CallToolResult -> plain Python value.""" + structured = getattr(result, "structuredContent", None) + if structured: + return structured + out = [] + for block in getattr(result, "content", []) or []: + text = getattr(block, "text", None) + if text is None: + out.append(block) + continue + try: + out.append(json.loads(text)) + except (ValueError, TypeError): + out.append(text) + if not out: + return None + return out[0] if len(out) == 1 else out + + +class _Worker: + """A single asyncio task (on a daemon thread) owns every MCP session. + + MCP's stdio sessions use anyio cancel scopes that must be entered and exited + in the same task, so all session work is funnelled through one worker task; + sync callers post commands over a queue and block on a thread-safe future. + """ + + def __init__(self): + self._loop = asyncio.new_event_loop() + threading.Thread(target=self._loop.run_forever, daemon=True).start() + self._queue: asyncio.Queue = self._call(asyncio.Queue) + asyncio.run_coroutine_threadsafe(self._run(), self._loop) + + def _call(self, fn): + fut: concurrent.futures.Future = concurrent.futures.Future() + self._loop.call_soon_threadsafe( + lambda: fut.set_result(fn()) if not fut.done() else None + ) + return fut.result() + + def _submit(self, action: str, payload): + fut: concurrent.futures.Future = concurrent.futures.Future() + self._loop.call_soon_threadsafe( + self._queue.put_nowait, (action, payload, fut) + ) + return fut.result() + + async def _run(self): + stack = AsyncExitStack() + sessions: Dict[str, ClientSession] = {} + stop_fut = None + try: + while True: + action, payload, fut = await self._queue.get() + if action == "stop": + stop_fut = fut + break + try: + if action == "open": + name, cmd, env = payload + params = StdioServerParameters( + command=cmd[0], args=cmd[1:], env=env + ) + read, write = await stack.enter_async_context( + stdio_client(params) + ) + session = await stack.enter_async_context( + ClientSession(read, write) + ) + await session.initialize() + sessions[name] = session + _set(fut, None) + elif action == "list": + _set(fut, await sessions[payload].list_tools()) + elif action == "call": + name, tool, args = payload + _set(fut, await sessions[name].call_tool(tool, args)) + except Exception as exc: + _set(fut, exc, error=True) + finally: + await stack.aclose() + if stop_fut is not None: + _set(stop_fut, None) + + def open(self, name, cmd, env): + self._submit("open", (name, cmd, env)) + + def list(self, name): + return self._submit("list", name) + + def call(self, name, tool, args): + return self._submit("call", (name, tool, args)) + + def stop(self): + try: + self._submit("stop", None) + finally: + self._loop.call_soon_threadsafe(self._loop.stop) + + +def _set(fut, value, error=False): + if fut.done(): + return + fut.set_exception(value) if error else fut.set_result(value) + + +class ToolUniverse: + """ToolUniverse-style facade over the AssetOpsBench MCP servers.""" + + def __init__(self, servers: Optional[Dict[str, List[str]]] = None, + env: Optional[dict] = None): + self.servers = servers or DEFAULT_SERVERS + self.env = {**os.environ, **(env or {})} + self._worker = _Worker() + self._connected: set = set() + self.all_tools: Dict[str, dict] = {} # qualified name -> spec + self._alias: Dict[str, str] = {} # bare name -> qualified + self._workflows: Dict[str, Callable] = {} + self._register_default_workflows() + + # -- context manager -- # + def __enter__(self): + return self + + def __exit__(self, *exc): + self.close() + + def close(self): + self._worker.stop() + + # -- step 2: load ------------------------------------------------------ # + def load_tools(self, servers: Optional[List[str]] = None) -> int: + """Connect to servers and discover their tools. Returns tool count.""" + names = list(servers) if servers else list(self.servers) + for server in names: + self._connect(server) + for t in self._worker.list(server).tools: + qualified = f"{server}.{t.name}" + self.all_tools[qualified] = { + "name": qualified, + "server": server, + "tool": t.name, + "description": t.description or "", + "parameters": getattr(t, "inputSchema", {}) or {}, + } + # Register a bare alias when it is unambiguous. + if t.name not in self._alias and t.name not in self.all_tools: + self._alias[t.name] = qualified + elif self._alias.get(t.name) not in (None, qualified): + _log.warning("Ambiguous tool name '%s'; use '%s'.", + t.name, qualified) + self._alias.pop(t.name, None) + return len(self.all_tools) + + # -- step 3: run ------------------------------------------------------- # + def run(self, query, arguments: Optional[dict] = None) -> Any: + """Execute a tool or workflow. + + ToolUniverse form: run({"name": "iot.sensors", "arguments": {...}}) + Shorthand: run("iot.sensors", {...}) + """ + if isinstance(query, dict): + name = query["name"] + args = query.get("arguments", {}) or {} + else: + name = query + args = arguments or {} + + if name in self._workflows: + return self._workflows[name](self, **args) + + server, tool = self._resolve(name) + result = self._worker.call(server, tool, args) + if getattr(result, "isError", False): + raise RuntimeError(f"Tool '{name}' failed: {_unwrap(result)}") + return _unwrap(result) + + # -- discovery --------------------------------------------------------- # + def find_tools(self, description: str, limit: int = 5) -> List[dict]: + """Keyword search over loaded tool names + descriptions.""" + terms = [w for w in description.lower().split() if w] + + def score(spec): + hay = f"{spec['name']} {spec['description']}".lower() + return sum(hay.count(t) for t in terms) + + ranked = sorted(self.all_tools.values(), key=score, reverse=True) + return [s for s in ranked if score(s) > 0][:limit] + + def list_tools(self, server: Optional[str] = None) -> List[str]: + """List loaded tool names, optionally for one server (lazy-connects).""" + if server is not None: + self._connect(server) + return [t.name for t in self._worker.list(server).tools] + return sorted(self.all_tools) + sorted(self._workflows) + + def tool_specification(self, name: str) -> dict: + return self.all_tools[self._qualify(name)] + + # -- workflows --------------------------------------------------------- # + def register_workflow(self, name: str, fn: Callable): + """Register a workflow callable fn(tu, **arguments) under `name`.""" + self._workflows[name] = fn + + def _register_default_workflows(self): + from . import workflows + for wf_name in getattr(workflows, "REGISTERED", []): + self._workflows[wf_name] = getattr(workflows, wf_name) + + # -- internals --------------------------------------------------------- # + def _connect(self, server: str): + if server not in self._connected: + if server not in self.servers: + raise KeyError( + f"Unknown server '{server}'. Known: {list(self.servers)}" + ) + self._worker.open(server, self.servers[server], self.env) + self._connected.add(server) + + def _qualify(self, name: str) -> str: + if name in self.all_tools: + return name + if name in self._alias: + return self._alias[name] + if "." in name: + return name + raise KeyError(f"Tool '{name}' not loaded. Call load_tools() first.") + + def _resolve(self, name: str): + qualified = self._qualify(name) + server, _, tool = qualified.partition(".") + if not tool: + raise ValueError(f"name must be '.', got '{name}'") + self._connect(server) + return server, tool + + +# Backwards-compatible alias. +ToolClient = ToolUniverse \ No newline at end of file diff --git a/src/mcphub/workflows.py b/src/mcphub/workflows.py new file mode 100644 index 00000000..7ea6be99 --- /dev/null +++ b/src/mcphub/workflows.py @@ -0,0 +1,66 @@ +""" +Composed workflows — plain functions that chain tool calls. + +Each takes a ``ToolUniverse`` instance and returns a dict. Names listed in +``REGISTERED`` are auto-registered so they run through the same entrypoint:: + + tu.run({"name": "chiller_triage", "arguments": {"asset_id": "Chiller 6"}}) + +To add one: write a function ``fn(tu, **arguments)`` and add its name to +``REGISTERED``. +""" + +REGISTERED = ["chiller_triage", "sensor_inventory_gap"] + + +def chiller_triage(tu, asset_id, site="MAIN", raise_work_order=True, priority="2"): + """Sensors -> failure modes -> mode/sensor mapping -> (optional) work order.""" + sensors = tu.run("iot.sensors", {"site_name": site, "asset_id": asset_id}) + failure_modes = tu.run("fmsr.get_failure_modes", {"asset_name": asset_id}) + mapping = tu.run("fmsr.get_failure_mode_sensor_mapping", { + "asset_name": asset_id, + "failure_modes": failure_modes, + "sensors": sensors, + }) + + out = { + "asset_id": asset_id, + "site_name": site, + "sensors": sensors, + "failure_modes": failure_modes, + "failure_mode_sensor_mapping": mapping, + } + if raise_work_order: + out["work_order"] = tu.run("wo.generate_work_order", { + "description": f"Investigate potential failure modes on {asset_id}", + "asset_num": asset_id, + "site_id": site, + "priority": priority, + "aob_source": "mcphub:chiller_triage", + }) + return out + + +def sensor_inventory_gap(tu, asset_id, site="MAIN"): + """Installed (registry) vs measured (telemetry) sensors for an asset.""" + installed = _names(tu.run("iot.asset_sensors", + {"site_name": site, "asset_id": asset_id})) + measured = _names(tu.run("iot.sensors", + {"site_name": site, "asset_id": asset_id})) + return { + "asset_id": asset_id, + "site_name": site, + "installed_not_streaming": sorted(set(installed) - set(measured)), + "streaming_not_in_registry": sorted(set(measured) - set(installed)), + } + + +def _names(value): + if isinstance(value, dict): + for key in ("sensors", "names", "data", "items"): + if key in value: + return _names(value[key]) + return list(value.keys()) + if isinstance(value, list): + return [v if isinstance(v, str) else v.get("name", str(v)) for v in value] + return [value] \ No newline at end of file From 75fadd2249a910201ece7938cf352488ce8c40e5 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Tue, 30 Jun 2026 22:40:36 -0400 Subject: [PATCH 2/4] revised code Signed-off-by: Dhaval Patel --- INSTRUCTIONS.md | 48 +++++++++++++++++++++++++++++ examples/quickstart_tooluniverse.py | 2 +- 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/INSTRUCTIONS.md b/INSTRUCTIONS.md index 21a21e3d..b58b1cbd 100644 --- a/INSTRUCTIONS.md +++ b/INSTRUCTIONS.md @@ -8,6 +8,7 @@ This directory contains the MCP servers and infrastructure for the AssetOpsBench - [Quick Start](#quick-start) - [Environment Variables](#environment-variables) - [MCP Servers](#mcp-servers) — full reference in [docs/mcp-servers.md](docs/mcp-servers.md) +- [Python client (mcphub)](#python-client-mcphub) - [Example queries](#example-queries) - [Agents](#agents) - [Observability](#observability) @@ -150,6 +151,53 @@ Tool signatures, required env vars, and how to launch a server directly: **[docs --- +## Python client (mcphub) + +`src/mcphub/` calls the MCP tools directly from Python — the same servers the +agents use, but without an LLM in the loop. It exposes a ToolUniverse-style +`load_tools → run` interface and adds no dependencies beyond the `mcp` client the +project already requires. + +​```python +from mcphub import ToolUniverse + +tu = ToolUniverse() # 1. init +tu.load_tools() # 2. connect + discover +result = tu.run({ # 3. run + "name": "iot.sensors", + "arguments": {"site_name": "MAIN", "asset_id": "Chiller 6"}, +}) +tu.close() +​``` + +- Tools are namespaced `.` (e.g. `iot.sensors`, + `wo.generate_work_order`); a bare name works when unambiguous. +- `load_tools(servers=["iot", "fmsr"])` limits the set; no args loads all six. +- Discovery: `tu.find_tools("failure mode")`, `tu.list_tools()`, + `tu.tool_specification("iot.sensors")`. +- Servers are spawned via `uv run -mcp-server` and inherit your `.env`. + Run inside the project (`uv run python ...`) or an activated venv. Override the + launch table with `ToolUniverse(servers={...})`. + +**Workflows** — multi-tool routines are plain functions registered under a name +and invoked through the same entrypoint: + +​```python +tu.run({"name": "chiller_triage", "arguments": {"asset_id": "Chiller 6"}}) +​``` + +Built in: `chiller_triage` (sensors → failure modes → mapping → work order) and +`sensor_inventory_gap` (installed vs measured sensors). Add your own in +`src/mcphub/workflows.py`. Runnable example: +`uv run python examples/quickstart_tooluniverse.py`. + +Reach for `mcphub` when you want deterministic, scripted tool orchestration +(pipelines, tests, data prep); use the agent runners when you want LLM-driven +planning. + +--- + + ## Example queries The CLI examples below use a `$query` shell variable so you can swap in any question without editing the commands. Pick one of these to get started: diff --git a/examples/quickstart_tooluniverse.py b/examples/quickstart_tooluniverse.py index 9d20dc31..131feeb3 100644 --- a/examples/quickstart_tooluniverse.py +++ b/examples/quickstart_tooluniverse.py @@ -4,7 +4,7 @@ docker compose -f src/couchdb/docker-compose.yaml up -d cp .env.public .env # WATSONX_* for fmsr Run: - uv run python examples/quickstart.py + uv run python examples/quickstart_tooluniverse.py """ import json From f702c13d45288b09804ffb0874b6637008d9ebb8 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Tue, 30 Jun 2026 22:47:30 -0400 Subject: [PATCH 3/4] revised code Signed-off-by: Dhaval Patel --- INSTRUCTIONS.md | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/INSTRUCTIONS.md b/INSTRUCTIONS.md index b58b1cbd..ba74416f 100644 --- a/INSTRUCTIONS.md +++ b/INSTRUCTIONS.md @@ -155,8 +155,7 @@ Tool signatures, required env vars, and how to launch a server directly: **[docs `src/mcphub/` calls the MCP tools directly from Python — the same servers the agents use, but without an LLM in the loop. It exposes a ToolUniverse-style -`load_tools → run` interface and adds no dependencies beyond the `mcp` client the -project already requires. +`load_tools → run` interface and adds no dependencies beyond the `mcp` client the project already requires. ​```python from mcphub import ToolUniverse @@ -202,7 +201,7 @@ planning. The CLI examples below use a `$query` shell variable so you can swap in any question without editing the commands. Pick one of these to get started: -```bash +```python # Simple single-server queries query="What sensors are on Chiller 6?" query="Is LSTM model supported in TSFM?" From abea0c3926cc9d227679da55938c5d2bef2e41d7 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Tue, 30 Jun 2026 23:09:55 -0400 Subject: [PATCH 4/4] revised Signed-off-by: Dhaval Patel --- INSTRUCTIONS.md | 22 +++++++++------------- 1 file changed, 9 insertions(+), 13 deletions(-) diff --git a/INSTRUCTIONS.md b/INSTRUCTIONS.md index ba74416f..2f7b6769 100644 --- a/INSTRUCTIONS.md +++ b/INSTRUCTIONS.md @@ -155,9 +155,10 @@ Tool signatures, required env vars, and how to launch a server directly: **[docs `src/mcphub/` calls the MCP tools directly from Python — the same servers the agents use, but without an LLM in the loop. It exposes a ToolUniverse-style -`load_tools → run` interface and adds no dependencies beyond the `mcp` client the project already requires. +`load_tools → run` interface and adds no dependencies beyond the `mcp` client the +project already requires. -​```python +```python from mcphub import ToolUniverse tu = ToolUniverse() # 1. init @@ -167,7 +168,7 @@ result = tu.run({ # 3. run "arguments": {"site_name": "MAIN", "asset_id": "Chiller 6"}, }) tu.close() -​``` +``` - Tools are namespaced `.` (e.g. `iot.sensors`, `wo.generate_work_order`); a bare name works when unambiguous. @@ -181,9 +182,9 @@ tu.close() **Workflows** — multi-tool routines are plain functions registered under a name and invoked through the same entrypoint: -​```python +```python tu.run({"name": "chiller_triage", "arguments": {"asset_id": "Chiller 6"}}) -​``` +``` Built in: `chiller_triage` (sensors → failure modes → mapping → work order) and `sensor_inventory_gap` (installed vs measured sensors). Add your own in @@ -196,12 +197,11 @@ planning. --- - ## Example queries The CLI examples below use a `$query` shell variable so you can swap in any question without editing the commands. Pick one of these to get started: -```python +```bash # Simple single-server queries query="What sensors are on Chiller 6?" query="Is LSTM model supported in TSFM?" @@ -428,12 +428,8 @@ uv run pytest src/ -k "integration" # only files / tests with "in ``` ┌────────────────────────────────────────────────────────────────────────┐ -│ agent/ │ -│ │ -│ PlanExecuteRunner ClaudeAgentRunner StirrupAgentRunner │ -│ OpenAIAgentRunner DeepAgentRunner OpenCodeAgentRunner │ -│ DirectLLMAgentRunner │ -│ │ +│ agent runners mcphub (Python client) │ +│ PlanExecute · Claude · OpenAI · Deep · Stirrup · OpenCode · DirectLLM │ └──────────────────────────────┬─────────────────────────────────────────┘ │ MCP protocol (stdio) ┌─────────────────────┼───────────┬──────────┬──────┬───────────┐