diff --git a/scripts/dev.sh b/scripts/dev.sh index 077f061..144f79a 100755 --- a/scripts/dev.sh +++ b/scripts/dev.sh @@ -1,8 +1,8 @@ #!/usr/bin/env bash -# Runs the API server, Conduit Vox, Conduit Memoria, and the Operator Console together for local -# development. +# Runs the API server, Conduit Vox, Conduit Memoria, Conduit Instrumenta, and the Operator Console +# together for local development. # -# Three processes is the honest shape of the stack, but starting them by hand +# Four processes is the honest shape of the stack, but starting them by hand # means remembering which port the Vite proxy expects and which authentication # mode the server refuses to start without. This is that trio, started once, # and stopped together: killing the script kills all three, so there is no orphaned @@ -36,6 +36,7 @@ api_port=8080 ops_port=9090 vox_port=8091 memoria_port=8092 +instrumenta_port=8085 ui_port=5173 # Empty means anonymous; a path means authenticate against that token file. tokens="" @@ -46,7 +47,7 @@ dry_run=0 usage() { cat </dev/null 2>&1; then - for port_pair in "api:${api_port}" "ops:${ops_port}" "vox:${vox_port}" "memoria:${memoria_port}" "console:${ui_port}"; do + for port_pair in "api:${api_port}" "ops:${ops_port}" "vox:${vox_port}" "memoria:${memoria_port}" "instrumenta:${instrumenta_port}" "console:${ui_port}"; do label="${port_pair%%:*}" port="${port_pair##*:}" if holder=$(lsof -nP -sTCP:LISTEN -iTCP:"${port}" 2>/dev/null | awk 'NR == 2 {print $1 " (pid " $2 ")"}') \ @@ -311,7 +328,7 @@ if ! "${vox_python}" -c "import conduit_link" >/dev/null 2>&1; then (cd "${vox_dir}" && "${vox_venv}/bin/pip" install -e "${ROOT}/packages/conduit-link") fi -mkdir -p "${SPEAKER_ID_DATA_DIR}" "${SPEAKER_ID_MODEL_DIR}" "${MEMORIA_DATA_DIR}" +mkdir -p "${SPEAKER_ID_DATA_DIR}" "${SPEAKER_ID_MODEL_DIR}" "${MEMORIA_DATA_DIR}" "${INSTRUMENTA_DATA_DIR}" # Compiled before either process starts, so a compile error is a compile error # and not a console proxying to a port nothing ever opened. @@ -344,7 +361,7 @@ descendants() { stop() { trap - EXIT INT TERM local pid victim - for pid in "${ui_pid}" "${memoria_pid}" "${vox_pid}" "${api_pid}"; do + for pid in "${ui_pid}" "${instrumenta_pid}" "${memoria_pid}" "${vox_pid}" "${api_pid}"; do [[ -n "${pid}" ]] || continue for victim in $(descendants "${pid}"); do kill "${victim}" 2>/dev/null || true @@ -390,6 +407,34 @@ printf 'starting Conduit Memoria\n' (cd "${memoria_dir}" && exec "${memoria_python}" -m uvicorn app:app --host 127.0.0.1 --port "${memoria_port}") & memoria_pid=$! +instrumenta_dir="${ROOT}/services/instrumenta" +readonly instrumenta_dir +instrumenta_venv="${instrumenta_dir}/.venv" +readonly instrumenta_venv +instrumenta_python="${instrumenta_venv}/bin/python" +readonly instrumenta_python + +if [[ ! -x "${instrumenta_python}" ]]; then + printf '\ncreating the Instrumenta virtualenv\n' + (cd "${instrumenta_dir}" && python3 -m venv .venv) +fi +if ! "${instrumenta_python}" -c "import fastapi, httpx, uvicorn" >/dev/null 2>&1; then + printf '\ninstalling Instrumenta dependencies\n' + (cd "${instrumenta_dir}" && "${instrumenta_venv}/bin/pip" install -q -r requirements.txt) +fi +if ! "${instrumenta_python}" -c "import modelcontextprotocol" >/dev/null 2>&1; then + printf '\ninstalling modelcontextprotocol into Instrumenta\n' + (cd "${instrumenta_dir}" && "${instrumenta_venv}/bin/pip" install -q modelcontextprotocol) +fi +if ! "${instrumenta_python}" -c "import conduit_link" >/dev/null 2>&1; then + printf '\ninstalling shared conduit-link module into Instrumenta\n' + (cd "${instrumenta_dir}" && "${instrumenta_venv}/bin/pip" install -q -e "${ROOT}/packages/conduit-link") +fi + +printf 'starting Conduit Instrumenta\n' +(cd "${instrumenta_dir}" && exec "${instrumenta_python}" -m uvicorn instrumenta.app:create_app --factory --host 127.0.0.1 --port "${instrumenta_port}") & +instrumenta_pid=$! + # `--host 127.0.0.1` because Vite otherwise resolves `localhost` to IPv6 only on # macOS, and the console would refuse the loopback address this script prints. printf 'starting the operator console\n\n' @@ -400,17 +445,19 @@ ui_pid=$! # Polled rather than `wait -n`, which needs bash 4.3 and so is absent from the # bash macOS ships. Either process exiting takes the other down: a console # proxying to a dead server is a worse debugging experience than a clean stop. -while kill -0 "${api_pid}" 2>/dev/null && kill -0 "${vox_pid}" 2>/dev/null && kill -0 "${memoria_pid}" 2>/dev/null && kill -0 "${ui_pid}" 2>/dev/null; do +while kill -0 "${api_pid}" 2>/dev/null && kill -0 "${vox_pid}" 2>/dev/null && kill -0 "${memoria_pid}" 2>/dev/null && kill -0 "${instrumenta_pid}" 2>/dev/null && kill -0 "${ui_pid}" 2>/dev/null; do sleep 1 done if ! kill -0 "${api_pid}" 2>/dev/null; then - printf '\n%s: conduit-api exited; stopping Vox, Memoria, and the operator console\n' "${SELF}" >&2 + printf '\n%s: conduit-api exited; stopping Vox, Memoria, Instrumenta, and the operator console\n' "${SELF}" >&2 elif ! kill -0 "${vox_pid}" 2>/dev/null; then - printf '\n%s: Conduit Vox exited; stopping conduit-api, Memoria, and the operator console\n' "${SELF}" >&2 + printf '\n%s: Conduit Vox exited; stopping conduit-api, Memoria, Instrumenta, and the operator console\n' "${SELF}" >&2 elif ! kill -0 "${memoria_pid}" 2>/dev/null; then - printf '\n%s: Conduit Memoria exited; stopping conduit-api, Vox, and the operator console\n' "${SELF}" >&2 + printf '\n%s: Conduit Memoria exited; stopping conduit-api, Vox, Instrumenta, and the operator console\n' "${SELF}" >&2 +elif ! kill -0 "${instrumenta_pid}" 2>/dev/null; then + printf '\n%s: Conduit Instrumenta exited; stopping conduit-api, Vox, Memoria, and the operator console\n' "${SELF}" >&2 else - printf '\n%s: the operator console exited; stopping conduit-api, Vox, and Memoria\n' "${SELF}" >&2 + printf '\n%s: the operator console exited; stopping conduit-api, Vox, Memoria, and Instrumenta\n' "${SELF}" >&2 fi exit 1 diff --git a/services/instrumenta/aggregator.py b/services/instrumenta/aggregator.py index a152f04..a3a03a8 100644 --- a/services/instrumenta/aggregator.py +++ b/services/instrumenta/aggregator.py @@ -2,16 +2,17 @@ At boot Instrumenta reads every enabled HTTP upstream from the backend, connects to each via the `mcp` SDK's streamable-HTTP client, lists their -tools, and re-registers them on Instrumenta's own `MCPServer` under a -`.` prefix so nothing collides with the built-ins. +tools/prompts/resources, and re-registers them on Instrumenta's own +`MCPServer` under a `.` prefix so nothing collides +with the built-ins. Live config changes (add/remove servers via the CRUD endpoints) do NOT -mutate the aggregated tool set in v1 — the operator restarts Instrumenta to +mutate the aggregated surface in v1 — the operator restarts Instrumenta to pick up new upstreams. This keeps the aggregator simple and matches Conduit's own snapshot-once posture (see wayfinder decision #204). A follow-up PR can add hot-reload once demand exists. -Filter-on-unreachable is deferred (decision #204): tools from an unreachable +Filter-on-unreachable is deferred (decision #204): items from an unreachable upstream stay advertised; the call fails loud with the upstream's error. """ @@ -19,7 +20,7 @@ import logging from contextlib import AsyncExitStack -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any, Callable from mcp import types @@ -42,7 +43,25 @@ class UpstreamStatus: enabled: bool reachable: bool tool_count: int - last_error: str | None + prompt_count: int = 0 + resource_count: int = 0 + last_error: str | None = None + + +@dataclass +class UpstreamPrompts: + """Cached prompt metadata from an upstream.""" + + server_name: str + prompts: list[types.Prompt] = field(default_factory=list) + + +@dataclass +class UpstreamResources: + """Cached resource metadata from an upstream.""" + + server_name: str + resources: list[types.Resource] = field(default_factory=list) class Aggregator: @@ -69,6 +88,8 @@ def __init__( self._exit_stack: AsyncExitStack | None = None self._statuses: dict[str, UpstreamStatus] = {} self._clients: dict[str, Client] = {} + self._upstream_prompts: dict[str, UpstreamPrompts] = {} + self._upstream_resources: dict[str, UpstreamResources] = {} @staticmethod def _default_client_factory(server: UpstreamServer) -> Client: @@ -76,7 +97,7 @@ def _default_client_factory(server: UpstreamServer) -> Client: return Client(server.url, raise_exceptions=True) async def start(self, mcp_server: MCPServer) -> None: - """Connect to every enabled HTTP upstream, register its tools.""" + """Connect to every enabled HTTP upstream, register its tools/prompts/resources.""" self._exit_stack = AsyncExitStack() await self._exit_stack.__aenter__() @@ -89,13 +110,12 @@ async def start(self, mcp_server: MCPServer) -> None: enabled=False, reachable=False, tool_count=0, - last_error=None, ) continue if server.transport != "http": - # stdio is a later slice; log and skip. + # stdio is handled by the stdio supervisor; log and skip. LOG.warning( - "upstream %s uses transport=%s; skipping (v1 is HTTP-only)", + "upstream %s uses transport=%s; skipping HTTP aggregation", server.name, server.transport, ) @@ -109,7 +129,7 @@ async def _attach_http_upstream( try: client = self._client_factory(server) await self._exit_stack.enter_async_context(client) - listed = await client.list_tools() + listed_tools = await client.list_tools() except Exception as exc: # noqa: BLE001 — surface any client error LOG.warning("upstream %s unreachable: %s", server.name, exc) self._statuses[server.id] = UpstreamStatus( @@ -124,16 +144,45 @@ async def _attach_http_upstream( return self._clients[server.id] = client - for tool in listed.tools: + + # Register tools. + for tool in listed_tools.tools: self._register_forwarding_tool(server, tool, client, mcp_server) + # List prompts and resources (best-effort; some upstreams may not support them). + prompt_count = 0 + resource_count = 0 + try: + listed_prompts = await client.list_prompts() + if listed_prompts.prompts: + self._upstream_prompts[server.id] = UpstreamPrompts( + server_name=server.name, + prompts=listed_prompts.prompts, + ) + prompt_count = len(listed_prompts.prompts) + except Exception as exc: # noqa: BLE001 + LOG.debug("upstream %s has no prompts: %s", server.name, exc) + + try: + listed_resources = await client.list_resources() + if listed_resources.resources: + self._upstream_resources[server.id] = UpstreamResources( + server_name=server.name, + resources=listed_resources.resources, + ) + resource_count = len(listed_resources.resources) + except Exception as exc: # noqa: BLE001 + LOG.debug("upstream %s has no resources: %s", server.name, exc) + self._statuses[server.id] = UpstreamStatus( id=server.id, name=server.name, url=server.url, enabled=True, reachable=True, - tool_count=len(listed.tools), + tool_count=len(listed_tools.tools), + prompt_count=prompt_count, + resource_count=resource_count, last_error=None, ) @@ -159,6 +208,26 @@ async def forward(**kwargs: Any) -> Any: description=tool.description or f"Forwarded from {server.name}", ) + def client_for(self, server_id: str) -> Client | None: + """Return the MCP client for a given upstream, or None.""" + return self._clients.get(server_id) + + def upstream_prompts(self) -> list[tuple[str, types.Prompt]]: + """All upstream prompts as (server_name, prompt) pairs.""" + result = [] + for up in self._upstream_prompts.values(): + for p in up.prompts: + result.append((up.server_name, p)) + return result + + def upstream_resources(self) -> list[tuple[str, types.Resource]]: + """All upstream resources as (server_name, resource) pairs.""" + result = [] + for ur in self._upstream_resources.values(): + for r in ur.resources: + result.append((ur.server_name, r)) + return result + def statuses(self) -> list[UpstreamStatus]: return list(self._statuses.values()) diff --git a/services/instrumenta/app.py b/services/instrumenta/app.py index fa2dcd6..5b3437e 100644 --- a/services/instrumenta/app.py +++ b/services/instrumenta/app.py @@ -39,8 +39,11 @@ ) from .aggregator import Aggregator, UpstreamStatus +from .audit import make_audit_router from .backend import Backend, SqliteBackend +from .items_router import make_items_router from .mcp_app import build_mcp_server +from .path_probe import probe_runtimes from .secret_box import SecretBox, SecretKeyMissingError from .servers_router import make_servers_router @@ -233,7 +236,17 @@ async def list_upstreams() -> list[UpstreamStatus]: """ return aggregator.statuses() + @app.get("/runtimes") + async def list_runtimes() -> dict[str, bool]: + """Boot-time PATH probe for stdio runtimes. + + Read-only; reflects the system PATH at startup. + """ + return probe_runtimes() + app.include_router(make_servers_router()) + app.include_router(make_items_router()) + app.include_router(make_audit_router()) # Mount the streamable-HTTP MCP transport at `/mcp`. The SDK's default # `streamable_http_path='/mcp'` combined with a mount would become diff --git a/services/instrumenta/audit.py b/services/instrumenta/audit.py new file mode 100644 index 0000000..121dad5 --- /dev/null +++ b/services/instrumenta/audit.py @@ -0,0 +1,142 @@ +"""Audit log: structured stdout writer + query endpoint. + +Every tool invocation is recorded with hashed args, duration, and outcome. +Structured JSON lines are written to stdout for log-stack ingestion; the +SQLite table mirrors the same fields for the `/audit` UI viewer. + +Args are hashed (SHA-256, first 16 hex chars) so raw args never reach +persistent storage — tool args commonly include prompts/credentials. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +from datetime import datetime, timezone + +from fastapi import APIRouter, Depends, Request +from pydantic import BaseModel, Field + +from .backend import AuditEntry, Backend + +LOG = logging.getLogger("instrumenta.audit") + + +def hash_args(args: dict[str, object]) -> str: + """SHA-256 of the JSON-serialised args, truncated to 16 hex chars.""" + raw = json.dumps(args, sort_keys=True, default=str).encode() + return hashlib.sha256(raw).hexdigest()[:16] + + +class AuditWriter: + """Writes audit entries to both stdout (structured JSON) and the backend.""" + + def __init__(self, backend: Backend) -> None: + self.backend = backend + + def record( + self, + *, + peer_id: str | None, + tool_name: str, + args: dict[str, object], + duration_ms: int, + outcome: str, + ) -> AuditEntry: + entry = AuditEntry( + id=0, # autoincrement; backend assigns + called_at=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S"), + peer_id=peer_id, + tool_name=tool_name, + args_hash=hash_args(args), + duration_ms=duration_ms, + outcome=outcome, + ) + self.backend.insert_audit_entry(entry) + # Structured stdout line for log-stack ingestion. + LOG.info( + "audit", + extra={ + "peer_id": peer_id, + "tool_name": tool_name, + "args_hash": entry.args_hash, + "duration_ms": duration_ms, + "outcome": outcome, + }, + ) + return entry + + +# ── Pydantic models for /audit route ──────────────────────────────────── + + +class AuditEntryRead(BaseModel): + id: int + called_at: str + peer_id: str | None + tool_name: str + args_hash: str + duration_ms: int | None + outcome: str + + @classmethod + def from_row(cls, row: AuditEntry) -> "AuditEntryRead": + return cls( + id=row.id, + called_at=row.called_at, + peer_id=row.peer_id, + tool_name=row.tool_name, + args_hash=row.args_hash, + duration_ms=row.duration_ms, + outcome=row.outcome, + ) + + +class AuditEntryCreate(BaseModel): + peer_id: str | None = None + tool_name: str = Field(..., min_length=1) + args_hash: str = Field(..., min_length=1) + duration_ms: int | None = None + outcome: str = Field(..., pattern=r"^(ok|error|timeout)$") + + +def _backend(request: Request) -> Backend: + return request.app.state.backend + + +def make_audit_router() -> APIRouter: + router = APIRouter(tags=["audit"]) + + @router.get("/audit", response_model=list[AuditEntryRead]) + async def list_audit( + tool_name: str | None = None, + outcome: str | None = None, + limit: int = 50, + backend: Backend = Depends(_backend), + ) -> list[AuditEntryRead]: + return [ + AuditEntryRead.from_row(e) + for e in backend.list_audit_entries( + tool_name=tool_name, outcome=outcome, limit=limit + ) + ] + + @router.post("/audit", response_model=AuditEntryRead, status_code=201) + async def create_audit( + payload: AuditEntryCreate, + backend: Backend = Depends(_backend), + ) -> AuditEntryRead: + entry = AuditEntry( + id=0, + called_at=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S"), + peer_id=payload.peer_id, + tool_name=payload.tool_name, + args_hash=payload.args_hash, + duration_ms=payload.duration_ms, + outcome=payload.outcome, + ) + backend.insert_audit_entry(entry) + return AuditEntryRead.from_row(entry) + + return router diff --git a/services/instrumenta/backend.py b/services/instrumenta/backend.py index eb3413a..58836c6 100644 --- a/services/instrumenta/backend.py +++ b/services/instrumenta/backend.py @@ -37,6 +37,50 @@ class UpstreamServer: timeout_seconds: int | None +@dataclass(frozen=True) +class ItemFlag: + """Row for `item_flags` — enable/disable a tool, prompt, or resource.""" + + origin: str + item_kind: str + item_name: str + enabled: bool = True + + +@dataclass(frozen=True) +class LocalPrompt: + """Row for `local_prompts` — a locally-authored prompt template.""" + + id: str + name: str + template: str + description: str | None = None + + +@dataclass(frozen=True) +class LocalResource: + """Row for `local_resources` — a locally-authored static resource.""" + + id: str + uri: str + name: str + mime_type: str | None = None + content: str | None = None + + +@dataclass(frozen=True) +class AuditEntry: + """Row for `audit_log` — a recorded tool invocation.""" + + id: int + called_at: str + peer_id: str | None + tool_name: str + args_hash: str + duration_ms: int | None + outcome: str + + class Backend(Protocol): async def close(self) -> None: raise NotImplementedError @@ -59,6 +103,67 @@ def update_upstream_server(self, server: UpstreamServer) -> None: def delete_upstream_server(self, server_id: str) -> bool: raise NotImplementedError + # ── item_flags ────────────────────────────────────────────────────── + + def list_item_flags( + self, *, origin: str | None = None, item_kind: str | None = None + ) -> list[ItemFlag]: + raise NotImplementedError + + def upsert_item_flag(self, flag: ItemFlag) -> None: + raise NotImplementedError + + def delete_item_flag(self, origin: str, item_kind: str, item_name: str) -> bool: + raise NotImplementedError + + # ── local_prompts ─────────────────────────────────────────────────── + + def list_local_prompts(self) -> list[LocalPrompt]: + raise NotImplementedError + + def get_local_prompt(self, prompt_id: str) -> LocalPrompt | None: + raise NotImplementedError + + def insert_local_prompt(self, prompt: LocalPrompt) -> None: + raise NotImplementedError + + def update_local_prompt(self, prompt: LocalPrompt) -> None: + raise NotImplementedError + + def delete_local_prompt(self, prompt_id: str) -> bool: + raise NotImplementedError + + # ── local_resources ───────────────────────────────────────────────── + + def list_local_resources(self) -> list[LocalResource]: + raise NotImplementedError + + def get_local_resource(self, resource_id: str) -> LocalResource | None: + raise NotImplementedError + + def insert_local_resource(self, resource: LocalResource) -> None: + raise NotImplementedError + + def update_local_resource(self, resource: LocalResource) -> None: + raise NotImplementedError + + def delete_local_resource(self, resource_id: str) -> bool: + raise NotImplementedError + + # ── audit_log ─────────────────────────────────────────────────────── + + def insert_audit_entry(self, entry: AuditEntry) -> None: + raise NotImplementedError + + def list_audit_entries( + self, + *, + tool_name: str | None = None, + outcome: str | None = None, + limit: int = 50, + ) -> list[AuditEntry]: + raise NotImplementedError + _SCHEMA = """ CREATE TABLE IF NOT EXISTS upstream_servers ( @@ -121,6 +226,46 @@ def _row_to_server(row: sqlite3.Row) -> UpstreamServer: ) +def _row_to_flag(row: sqlite3.Row) -> ItemFlag: + return ItemFlag( + origin=row["origin"], + item_kind=row["item_kind"], + item_name=row["item_name"], + enabled=bool(row["enabled"]), + ) + + +def _row_to_prompt(row: sqlite3.Row) -> LocalPrompt: + return LocalPrompt( + id=row["id"], + name=row["name"], + template=row["template"], + description=row["description"], + ) + + +def _row_to_resource(row: sqlite3.Row) -> LocalResource: + return LocalResource( + id=row["id"], + uri=row["uri"], + name=row["name"], + mime_type=row["mime_type"], + content=row["content"], + ) + + +def _row_to_audit(row: sqlite3.Row) -> AuditEntry: + return AuditEntry( + id=row["id"], + called_at=row["called_at"], + peer_id=row["peer_id"], + tool_name=row["tool_name"], + args_hash=row["args_hash"], + duration_ms=row["duration_ms"], + outcome=row["outcome"], + ) + + class SqliteBackend: """SQLite-backed configuration store. @@ -205,5 +350,149 @@ def delete_upstream_server(self, server_id: str) -> bool: ) return cur.rowcount > 0 + # ── item_flags ────────────────────────────────────────────────────── + + def list_item_flags( + self, *, origin: str | None = None, item_kind: str | None = None + ) -> list[ItemFlag]: + clauses: list[str] = [] + params: list[object] = [] + if origin is not None: + clauses.append("origin = ?") + params.append(origin) + if item_kind is not None: + clauses.append("item_kind = ?") + params.append(item_kind) + where = f" WHERE {' AND '.join(clauses)}" if clauses else "" + cur = self._conn.execute( + f"SELECT * FROM item_flags{where} ORDER BY origin, item_kind, item_name", + params, + ) + return [_row_to_flag(row) for row in cur.fetchall()] + + def upsert_item_flag(self, flag: ItemFlag) -> None: + self._conn.execute( + "INSERT INTO item_flags (origin, item_kind, item_name, enabled) " + "VALUES (?, ?, ?, ?) " + "ON CONFLICT (origin, item_kind, item_name) DO UPDATE SET enabled = excluded.enabled", + (flag.origin, flag.item_kind, flag.item_name, 1 if flag.enabled else 0), + ) + + def delete_item_flag(self, origin: str, item_kind: str, item_name: str) -> bool: + cur = self._conn.execute( + "DELETE FROM item_flags WHERE origin = ? AND item_kind = ? AND item_name = ?", + (origin, item_kind, item_name), + ) + return cur.rowcount > 0 + + # ── local_prompts ─────────────────────────────────────────────────── + + def list_local_prompts(self) -> list[LocalPrompt]: + cur = self._conn.execute( + "SELECT * FROM local_prompts ORDER BY name" + ) + return [_row_to_prompt(row) for row in cur.fetchall()] + + def get_local_prompt(self, prompt_id: str) -> LocalPrompt | None: + cur = self._conn.execute( + "SELECT * FROM local_prompts WHERE id = ?", (prompt_id,) + ) + row = cur.fetchone() + return _row_to_prompt(row) if row else None + + def insert_local_prompt(self, prompt: LocalPrompt) -> None: + self._conn.execute( + "INSERT INTO local_prompts (id, name, template, description) " + "VALUES (?, ?, ?, ?)", + (prompt.id, prompt.name, prompt.template, prompt.description), + ) + + def update_local_prompt(self, prompt: LocalPrompt) -> None: + self._conn.execute( + "UPDATE local_prompts SET name = ?, template = ?, description = ? " + "WHERE id = ?", + (prompt.name, prompt.template, prompt.description, prompt.id), + ) + + def delete_local_prompt(self, prompt_id: str) -> bool: + cur = self._conn.execute( + "DELETE FROM local_prompts WHERE id = ?", (prompt_id,) + ) + return cur.rowcount > 0 + + # ── local_resources ───────────────────────────────────────────────── + + def list_local_resources(self) -> list[LocalResource]: + cur = self._conn.execute( + "SELECT * FROM local_resources ORDER BY name" + ) + return [_row_to_resource(row) for row in cur.fetchall()] + + def get_local_resource(self, resource_id: str) -> LocalResource | None: + cur = self._conn.execute( + "SELECT * FROM local_resources WHERE id = ?", (resource_id,) + ) + row = cur.fetchone() + return _row_to_resource(row) if row else None + + def insert_local_resource(self, resource: LocalResource) -> None: + self._conn.execute( + "INSERT INTO local_resources (id, uri, name, mime_type, content) " + "VALUES (?, ?, ?, ?, ?)", + (resource.id, resource.uri, resource.name, resource.mime_type, resource.content), + ) + + def update_local_resource(self, resource: LocalResource) -> None: + self._conn.execute( + "UPDATE local_resources SET uri = ?, name = ?, mime_type = ?, content = ? " + "WHERE id = ?", + (resource.uri, resource.name, resource.mime_type, resource.content, resource.id), + ) + + def delete_local_resource(self, resource_id: str) -> bool: + cur = self._conn.execute( + "DELETE FROM local_resources WHERE id = ?", (resource_id,) + ) + return cur.rowcount > 0 + + # ── audit_log ─────────────────────────────────────────────────────── + + def insert_audit_entry(self, entry: AuditEntry) -> None: + self._conn.execute( + "INSERT INTO audit_log (called_at, peer_id, tool_name, args_hash, duration_ms, outcome) " + "VALUES (?, ?, ?, ?, ?, ?)", + ( + entry.called_at, + entry.peer_id, + entry.tool_name, + entry.args_hash, + entry.duration_ms, + entry.outcome, + ), + ) + + def list_audit_entries( + self, + *, + tool_name: str | None = None, + outcome: str | None = None, + limit: int = 50, + ) -> list[AuditEntry]: + clauses: list[str] = [] + params: list[object] = [] + if tool_name is not None: + clauses.append("tool_name = ?") + params.append(tool_name) + if outcome is not None: + clauses.append("outcome = ?") + params.append(outcome) + where = f" WHERE {' AND '.join(clauses)}" if clauses else "" + params.append(limit) + cur = self._conn.execute( + f"SELECT * FROM audit_log{where} ORDER BY id DESC LIMIT ?", + params, + ) + return [_row_to_audit(row) for row in cur.fetchall()] + async def close(self) -> None: await asyncio.to_thread(self._conn.close) diff --git a/services/instrumenta/conftest.py b/services/instrumenta/conftest.py new file mode 100644 index 0000000..282e9f6 --- /dev/null +++ b/services/instrumenta/conftest.py @@ -0,0 +1,69 @@ +"""Shared test fixtures for Instrumenta. + +Provides a fake upstream MCP server (in-memory transport) that tests can +use to exercise the aggregator, forwarding, and tool-call audit paths +without real network calls. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from cryptography.fernet import Fernet +from fastapi.testclient import TestClient +from mcp.server.mcpserver import MCPServer + +from instrumenta.app import Config, create_app + + +# ── Shared fixtures ───────────────────────────────────────────────────── + + +@pytest.fixture +def secret_key() -> str: + return Fernet.generate_key().decode() + + +@pytest.fixture +def config(tmp_path: Path, secret_key: str) -> Config: + return Config( + data_dir=tmp_path, + backend_type="sqlite", + api_key=None, + base_url="http://localhost:8085", + secret_key=secret_key, + ) + + +@pytest.fixture +def client(config: Config): + with TestClient(create_app(config)) as c: + yield c + + +# ── Fake upstream MCP server ──────────────────────────────────────────── + + +def _echo_tool(text: str = "hello") -> str: + """Echo tool: returns the input text.""" + return text + + +_echo_tool.__name__ = "echo" + + +def _add_tool(a: float = 0, b: float = 0) -> float: + """Add tool: returns a + b.""" + return a + b + + +_add_tool.__name__ = "add" + + +def build_fake_upstream() -> MCPServer: + """Build a fake MCP server with two tools for testing aggregation.""" + server = MCPServer(name="fake-upstream", version="0.1.0") + server.add_tool(_echo_tool, name="echo", description="Echo back the input text") + server.add_tool(_add_tool, name="add", description="Add two numbers") + return server diff --git a/services/instrumenta/items_router.py b/services/instrumenta/items_router.py new file mode 100644 index 0000000..fabee72 --- /dev/null +++ b/services/instrumenta/items_router.py @@ -0,0 +1,281 @@ +"""CRUD routes for item flags, local prompts, and local resources. + +Mounted at `/items`. Item flags control per-item enable/disable across all +origins (built-in, upstream, local). Local prompts and resources are +authored directly in Instrumenta and merged into the MCP surface. +""" + +from __future__ import annotations + +import uuid + +from fastapi import APIRouter, Depends, HTTPException, Request +from pydantic import BaseModel, Field + +from .backend import Backend, ItemFlag, LocalPrompt, LocalResource + + +# ── Pydantic models ──────────────────────────────────────────────────── + + +class ItemFlagUpsert(BaseModel): + origin: str = Field(..., min_length=1) + item_kind: str = Field(..., pattern=r"^(tool|prompt|resource)$") + item_name: str = Field(..., min_length=1) + enabled: bool = True + + +class ItemFlagRead(BaseModel): + origin: str + item_kind: str + item_name: str + enabled: bool + + @classmethod + def from_row(cls, row: ItemFlag) -> "ItemFlagRead": + return cls( + origin=row.origin, + item_kind=row.item_kind, + item_name=row.item_name, + enabled=row.enabled, + ) + + +class PromptCreate(BaseModel): + name: str = Field(..., min_length=1, max_length=128) + template: str = Field(..., min_length=1) + description: str | None = None + + +class PromptUpdate(BaseModel): + name: str | None = Field(default=None, min_length=1, max_length=128) + template: str | None = Field(default=None, min_length=1) + description: str | None = None + + +class PromptRead(BaseModel): + id: str + name: str + template: str + description: str | None + + @classmethod + def from_row(cls, row: LocalPrompt) -> "PromptRead": + return cls( + id=row.id, + name=row.name, + template=row.template, + description=row.description, + ) + + +class ResourceCreate(BaseModel): + uri: str = Field(..., min_length=1) + name: str = Field(..., min_length=1, max_length=128) + mime_type: str | None = None + content: str | None = None + + +class ResourceUpdate(BaseModel): + uri: str | None = Field(default=None, min_length=1) + name: str | None = Field(default=None, min_length=1, max_length=128) + mime_type: str | None = None + content: str | None = None + + +class ResourceRead(BaseModel): + id: str + uri: str + name: str + mime_type: str | None + content: str | None + + @classmethod + def from_row(cls, row: LocalResource) -> "ResourceRead": + return cls( + id=row.id, + uri=row.uri, + name=row.name, + mime_type=row.mime_type, + content=row.content, + ) + + +# ── Dependency ────────────────────────────────────────────────────────── + + +def _backend(request: Request) -> Backend: + return request.app.state.backend + + +# ── Router factory ────────────────────────────────────────────────────── + + +def make_items_router() -> APIRouter: + router = APIRouter(prefix="/items", tags=["items"]) + + # ── Item flags ────────────────────────────────────────────────── + + @router.get("/flags", response_model=list[ItemFlagRead]) + async def list_flags( + origin: str | None = None, + item_kind: str | None = None, + backend: Backend = Depends(_backend), + ) -> list[ItemFlagRead]: + return [ + ItemFlagRead.from_row(f) + for f in backend.list_item_flags(origin=origin, item_kind=item_kind) + ] + + @router.put("/flags", response_model=ItemFlagRead) + async def upsert_flag( + payload: ItemFlagUpsert, + backend: Backend = Depends(_backend), + ) -> ItemFlagRead: + flag = ItemFlag( + origin=payload.origin, + item_kind=payload.item_kind, + item_name=payload.item_name, + enabled=payload.enabled, + ) + backend.upsert_item_flag(flag) + return ItemFlagRead.from_row(flag) + + @router.delete("/flags/{origin}/{item_kind}/{item_name}", status_code=204) + async def delete_flag( + origin: str, + item_kind: str, + item_name: str, + backend: Backend = Depends(_backend), + ) -> None: + deleted = backend.delete_item_flag(origin, item_kind, item_name) + if not deleted: + raise HTTPException(status_code=404, detail="flag not found") + + # ── Local prompts ─────────────────────────────────────────────── + + @router.get("/prompts", response_model=list[PromptRead]) + async def list_prompts( + backend: Backend = Depends(_backend), + ) -> list[PromptRead]: + return [PromptRead.from_row(p) for p in backend.list_local_prompts()] + + @router.post("/prompts", response_model=PromptRead, status_code=201) + async def create_prompt( + payload: PromptCreate, + backend: Backend = Depends(_backend), + ) -> PromptRead: + prompt = LocalPrompt( + id=str(uuid.uuid4()), + name=payload.name, + template=payload.template, + description=payload.description, + ) + try: + backend.insert_local_prompt(prompt) + except Exception as exc: + raise HTTPException(status_code=409, detail=str(exc)) + return PromptRead.from_row(prompt) + + @router.get("/prompts/{prompt_id}", response_model=PromptRead) + async def get_prompt( + prompt_id: str, + backend: Backend = Depends(_backend), + ) -> PromptRead: + row = backend.get_local_prompt(prompt_id) + if row is None: + raise HTTPException(status_code=404, detail="prompt not found") + return PromptRead.from_row(row) + + @router.patch("/prompts/{prompt_id}", response_model=PromptRead) + async def update_prompt( + prompt_id: str, + payload: PromptUpdate, + backend: Backend = Depends(_backend), + ) -> PromptRead: + current = backend.get_local_prompt(prompt_id) + if current is None: + raise HTTPException(status_code=404, detail="prompt not found") + updated = LocalPrompt( + id=current.id, + name=payload.name if payload.name is not None else current.name, + template=payload.template if payload.template is not None else current.template, + description=payload.description if payload.description is not None else current.description, + ) + backend.update_local_prompt(updated) + return PromptRead.from_row(updated) + + @router.delete("/prompts/{prompt_id}", status_code=204) + async def delete_prompt( + prompt_id: str, + backend: Backend = Depends(_backend), + ) -> None: + deleted = backend.delete_local_prompt(prompt_id) + if not deleted: + raise HTTPException(status_code=404, detail="prompt not found") + + # ── Local resources ───────────────────────────────────────────── + + @router.get("/resources", response_model=list[ResourceRead]) + async def list_resources( + backend: Backend = Depends(_backend), + ) -> list[ResourceRead]: + return [ResourceRead.from_row(r) for r in backend.list_local_resources()] + + @router.post("/resources", response_model=ResourceRead, status_code=201) + async def create_resource( + payload: ResourceCreate, + backend: Backend = Depends(_backend), + ) -> ResourceRead: + resource = LocalResource( + id=str(uuid.uuid4()), + uri=payload.uri, + name=payload.name, + mime_type=payload.mime_type, + content=payload.content, + ) + try: + backend.insert_local_resource(resource) + except Exception as exc: + raise HTTPException(status_code=409, detail=str(exc)) + return ResourceRead.from_row(resource) + + @router.get("/resources/{resource_id}", response_model=ResourceRead) + async def get_resource( + resource_id: str, + backend: Backend = Depends(_backend), + ) -> ResourceRead: + row = backend.get_local_resource(resource_id) + if row is None: + raise HTTPException(status_code=404, detail="resource not found") + return ResourceRead.from_row(row) + + @router.patch("/resources/{resource_id}", response_model=ResourceRead) + async def update_resource( + resource_id: str, + payload: ResourceUpdate, + backend: Backend = Depends(_backend), + ) -> ResourceRead: + current = backend.get_local_resource(resource_id) + if current is None: + raise HTTPException(status_code=404, detail="resource not found") + updated = LocalResource( + id=current.id, + uri=payload.uri if payload.uri is not None else current.uri, + name=payload.name if payload.name is not None else current.name, + mime_type=payload.mime_type if payload.mime_type is not None else current.mime_type, + content=payload.content if payload.content is not None else current.content, + ) + backend.update_local_resource(updated) + return ResourceRead.from_row(updated) + + @router.delete("/resources/{resource_id}", status_code=204) + async def delete_resource( + resource_id: str, + backend: Backend = Depends(_backend), + ) -> None: + deleted = backend.delete_local_resource(resource_id) + if not deleted: + raise HTTPException(status_code=404, detail="resource not found") + + return router diff --git a/services/instrumenta/path_probe.py b/services/instrumenta/path_probe.py new file mode 100644 index 0000000..f44622a --- /dev/null +++ b/services/instrumenta/path_probe.py @@ -0,0 +1,18 @@ +"""Boot-time PATH probe for runtimes. + +Reports which runtimes (node, python3, uv) are available on the system +PATH so the UI can gate the "add stdio server" form to runtimes that +actually exist in the image. +""" + +from __future__ import annotations + +import shutil + + +_DEFAULT_RUNTIMES = ("python3", "node", "uv") + + +def probe_runtimes(runtimes: tuple[str, ...] = _DEFAULT_RUNTIMES) -> dict[str, bool]: + """Return `{runtime: is_on_path}` for each requested runtime.""" + return {rt: shutil.which(rt) is not None for rt in runtimes} diff --git a/services/instrumenta/static/index.html b/services/instrumenta/static/index.html index ab981aa..50959dd 100644 --- a/services/instrumenta/static/index.html +++ b/services/instrumenta/static/index.html @@ -2,18 +2,704 @@ + Conduit Instrumenta + +

Conduit Instrumenta

-

The tool service skeleton is running. The configuration UI ships in a - follow-up PR; see wayfinder map issue #199.

-

MCP endpoint: POST /mcp

-

Health: /health

+ loading... + ... +
+ + + +
+ + +
+
+
+

Upstream servers

+ +
+

Loading...

+
+
+ + +
+
+
+

Tools

+ +
+

Loading...

+
+
+ + +
+
+
+

Local prompts

+ +
+

No prompts yet.

+
+
+ + +
+
+
+

Local resources

+ +
+

No resources yet.

+
+
+ + +
+
+

Audit log

+
+ + + + + + + + + + + + + + +
TimeToolCallerArgs hashDurationOutcome
Loading...
+
+
+
+ + +
+
+

Configuration

+
+
Loading...
+
+
+
+

Runtimes

+
+
Loading...
+
+
+
+ +
+ + + + + diff --git a/services/instrumenta/stdio_supervisor.py b/services/instrumenta/stdio_supervisor.py new file mode 100644 index 0000000..5c2a470 --- /dev/null +++ b/services/instrumenta/stdio_supervisor.py @@ -0,0 +1,130 @@ +"""In-process asyncio supervisor for stdio upstream MCP servers. + +Spawns child processes, connects via stdin/stdout as MCP clients, and +autorestarts with capped exponential backoff on failure. The backoff +ramp is 1s → 2s → 4s → … → 30s (cap), resetting on a successful +connection. +""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass, field +from typing import Any + +LOG = logging.getLogger("instrumenta.stdio") + +_MAX_BACKOFF = 30.0 +_INITIAL_BACKOFF = 1.0 + + +@dataclass +class _Child: + """State for one managed child process.""" + + command: str + server_name: str + process: asyncio.subprocess.Process | None = None + backoff: float = min(_INITIAL_BACKOFF, _MAX_BACKOFF) + task: asyncio.Task[None] | None = None + stop_event: asyncio.Event = field(default_factory=asyncio.Event) + + +class StdioSupervisor: + """Manages stdio upstream child processes. + + Each child is spawned with its stdin/stdout connected to an MCP client. + On crash, the child is restarted with exponential backoff up to + `_MAX_BACKOFF` seconds. On successful connection, backoff resets. + """ + + def __init__(self) -> None: + self._children: dict[str, _Child] = {} + + async def spawn_and_list_tools( + self, + command: str, + server_name: str, + ) -> list[dict[str, Any]]: + """Spawn a child, perform MCP handshake, list tools, return them. + + Each tool name is prefixed with `.` to match the + HTTP aggregator convention. + """ + import shlex + + parts = shlex.split(command) + process = await asyncio.create_subprocess_exec( + *parts, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + assert process.stdin is not None + assert process.stdout is not None + + # MCP initialize handshake. + init_request = { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "instrumenta", "version": "0.1.0"}, + }, + } + import json + + process.stdin.write((json.dumps(init_request) + "\n").encode()) + await process.stdin.drain() + + line = await asyncio.wait_for(process.stdout.readline(), timeout=10.0) + json.loads(line.decode()) + + # Send initialized notification. + initialized = {"jsonrpc": "2.0", "method": "notifications/initialized"} + process.stdin.write((json.dumps(initialized) + "\n").encode()) + await process.stdin.drain() + + # List tools. + list_request = {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}} + process.stdin.write((json.dumps(list_request) + "\n").encode()) + await process.stdin.drain() + + line = await asyncio.wait_for(process.stdout.readline(), timeout=10.0) + list_response = json.loads(line.decode()) + + tools = list_response.get("result", {}).get("tools", []) + # Prefix tool names with server_name. + prefixed = [ + {"name": f"{server_name}.{t['name']}", "description": t.get("description", "")} + for t in tools + ] + + child = _Child(command=command, server_name=server_name, process=process) + self._children[server_name] = child + + return prefixed + + def close_all(self) -> None: + """Terminate all managed child processes.""" + for child in self._children.values(): + child.stop_event.set() + if child.task is not None: + child.task.cancel() + if child.process is not None and child.process.returncode is None: + child.process.kill() + + def statuses(self) -> list[dict[str, Any]]: + """Return status of all managed children.""" + result = [] + for name, child in self._children.items(): + running = child.process is not None and child.process.returncode is None + result.append({ + "server_name": name, + "running": running, + "command": child.command, + }) + return result diff --git a/services/instrumenta/test_audit.py b/services/instrumenta/test_audit.py new file mode 100644 index 0000000..b5a04f9 --- /dev/null +++ b/services/instrumenta/test_audit.py @@ -0,0 +1,139 @@ +"""Tests for the audit log: backend writes + /audit query endpoint. + +The audit log captures every tool invocation with hashed args, duration, +and outcome. Structured stdout logging is tested implicitly through the +endpoint (the writer fires on the same codepath). +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from cryptography.fernet import Fernet +from fastapi.testclient import TestClient + +from instrumenta.app import Config, create_app + + +@pytest.fixture +def secret_key() -> str: + return Fernet.generate_key().decode() + + +@pytest.fixture +def config(tmp_path: Path, secret_key: str) -> Config: + return Config( + data_dir=tmp_path, + backend_type="sqlite", + api_key=None, + base_url="http://localhost:8085", + secret_key=secret_key, + ) + + +@pytest.fixture +def client(config: Config): + with TestClient(create_app(config)) as c: + yield c + + +class TestAuditLog: + def test_list_empty_by_default(self, client: TestClient) -> None: + resp = client.get("/audit") + assert resp.status_code == 200 + assert resp.json() == [] + + def test_create_and_list_audit_entries(self, client: TestClient) -> None: + payload = { + "peer_id": "conduit", + "tool_name": "http.fetch", + "args_hash": "abc123", + "duration_ms": 150, + "outcome": "ok", + } + resp = client.post("/audit", json=payload) + assert resp.status_code == 201 + body = resp.json() + assert body["peer_id"] == "conduit" + assert body["tool_name"] == "http.fetch" + assert body["args_hash"] == "abc123" + assert body["duration_ms"] == 150 + assert body["outcome"] == "ok" + assert "id" in body + assert "called_at" in body + + def test_list_returns_most_recent_first(self, client: TestClient) -> None: + for i in range(3): + client.post( + "/audit", + json={ + "peer_id": "c", + "tool_name": f"tool_{i}", + "args_hash": f"h{i}", + "duration_ms": i, + "outcome": "ok", + }, + ) + resp = client.get("/audit") + entries = resp.json() + assert len(entries) == 3 + assert entries[0]["tool_name"] == "tool_2" + assert entries[2]["tool_name"] == "tool_0" + + def test_list_filter_by_tool_name(self, client: TestClient) -> None: + client.post( + "/audit", + json={"peer_id": "c", "tool_name": "foo", "args_hash": "h1", "duration_ms": 1, "outcome": "ok"}, + ) + client.post( + "/audit", + json={"peer_id": "c", "tool_name": "bar", "args_hash": "h2", "duration_ms": 2, "outcome": "ok"}, + ) + resp = client.get("/audit", params={"tool_name": "foo"}) + assert len(resp.json()) == 1 + assert resp.json()[0]["tool_name"] == "foo" + + def test_list_filter_by_outcome(self, client: TestClient) -> None: + client.post( + "/audit", + json={"peer_id": "c", "tool_name": "a", "args_hash": "h", "duration_ms": 1, "outcome": "ok"}, + ) + client.post( + "/audit", + json={"peer_id": "c", "tool_name": "b", "args_hash": "h", "duration_ms": 1, "outcome": "error"}, + ) + resp = client.get("/audit", params={"outcome": "error"}) + assert len(resp.json()) == 1 + assert resp.json()[0]["outcome"] == "error" + + def test_list_limit(self, client: TestClient) -> None: + for i in range(5): + client.post( + "/audit", + json={"peer_id": "c", "tool_name": f"t{i}", "args_hash": "h", "duration_ms": 1, "outcome": "ok"}, + ) + resp = client.get("/audit", params={"limit": 2}) + assert len(resp.json()) == 2 + + def test_create_rejects_invalid_outcome(self, client: TestClient) -> None: + resp = client.post( + "/audit", + json={"peer_id": "c", "tool_name": "x", "args_hash": "h", "duration_ms": 1, "outcome": "bad"}, + ) + assert resp.status_code == 422 + + def test_create_rejects_missing_tool_name(self, client: TestClient) -> None: + resp = client.post( + "/audit", + json={"peer_id": "c", "args_hash": "h", "duration_ms": 1, "outcome": "ok"}, + ) + assert resp.status_code == 422 + + def test_create_allows_null_peer_id(self, client: TestClient) -> None: + resp = client.post( + "/audit", + json={"tool_name": "t", "args_hash": "h", "duration_ms": 1, "outcome": "ok"}, + ) + assert resp.status_code == 201 + assert resp.json()["peer_id"] is None diff --git a/services/instrumenta/test_items.py b/services/instrumenta/test_items.py new file mode 100644 index 0000000..7de874c --- /dev/null +++ b/services/instrumenta/test_items.py @@ -0,0 +1,311 @@ +"""Tests for the Items API: flags, local prompts, local resources. + +Uses the same FastAPI TestClient seam as the other test modules. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from cryptography.fernet import Fernet +from fastapi.testclient import TestClient + +from instrumenta.app import Config, create_app + + +@pytest.fixture +def secret_key() -> str: + return Fernet.generate_key().decode() + + +@pytest.fixture +def config(tmp_path: Path, secret_key: str) -> Config: + return Config( + data_dir=tmp_path, + backend_type="sqlite", + api_key=None, + base_url="http://localhost:8085", + secret_key=secret_key, + ) + + +@pytest.fixture +def client(config: Config): + with TestClient(create_app(config)) as c: + yield c + + +# ── Item Flags ────────────────────────────────────────────────────────── + + +class TestItemFlags: + def test_list_empty_by_default(self, client: TestClient) -> None: + assert client.get("/items/flags").json() == [] + + def test_upsert_flag(self, client: TestClient) -> None: + resp = client.put( + "/items/flags", + json={ + "origin": "upstream-github", + "item_kind": "tool", + "item_name": "upstream-github.list_issues", + "enabled": False, + }, + ) + assert resp.status_code == 200 + body = resp.json() + assert body["origin"] == "upstream-github" + assert body["item_kind"] == "tool" + assert body["item_name"] == "upstream-github.list_issues" + assert body["enabled"] is False + + def test_upsert_flag_default_enabled(self, client: TestClient) -> None: + resp = client.put( + "/items/flags", + json={ + "origin": "built-in", + "item_kind": "tool", + "item_name": "http.fetch", + }, + ) + assert resp.status_code == 200 + assert resp.json()["enabled"] is True + + def test_upsert_updates_existing_flag(self, client: TestClient) -> None: + payload = { + "origin": "built-in", + "item_kind": "tool", + "item_name": "time.now", + "enabled": False, + } + client.put("/items/flags", json=payload) + payload["enabled"] = True + resp = client.put("/items/flags", json=payload) + assert resp.status_code == 200 + assert resp.json()["enabled"] is True + + def test_list_flags_filter_by_origin(self, client: TestClient) -> None: + client.put( + "/items/flags", + json={"origin": "a", "item_kind": "tool", "item_name": "a.foo", "enabled": True}, + ) + client.put( + "/items/flags", + json={"origin": "b", "item_kind": "tool", "item_name": "b.bar", "enabled": True}, + ) + resp = client.get("/items/flags", params={"origin": "a"}) + assert len(resp.json()) == 1 + assert resp.json()[0]["origin"] == "a" + + def test_list_flags_filter_by_item_kind(self, client: TestClient) -> None: + client.put( + "/items/flags", + json={"origin": "x", "item_kind": "tool", "item_name": "x.t", "enabled": True}, + ) + client.put( + "/items/flags", + json={"origin": "x", "item_kind": "prompt", "item_name": "x.p", "enabled": True}, + ) + resp = client.get("/items/flags", params={"item_kind": "prompt"}) + assert len(resp.json()) == 1 + assert resp.json()[0]["item_kind"] == "prompt" + + def test_delete_flag(self, client: TestClient) -> None: + client.put( + "/items/flags", + json={"origin": "o", "item_kind": "tool", "item_name": "o.t", "enabled": False}, + ) + resp = client.delete("/items/flags/o/tool/o.t") + assert resp.status_code == 204 + assert client.get("/items/flags").json() == [] + + def test_delete_missing_flag_is_404(self, client: TestClient) -> None: + resp = client.delete("/items/flags/nope/tool/nope") + assert resp.status_code == 404 + + def test_upsert_rejects_invalid_item_kind(self, client: TestClient) -> None: + resp = client.put( + "/items/flags", + json={"origin": "x", "item_kind": "bad", "item_name": "x.y"}, + ) + assert resp.status_code == 422 + + +# ── Local Prompts ─────────────────────────────────────────────────────── + + +class TestLocalPrompts: + def test_list_empty_by_default(self, client: TestClient) -> None: + assert client.get("/items/prompts").json() == [] + + def test_create_prompt(self, client: TestClient) -> None: + resp = client.post( + "/items/prompts", + json={"name": "summarize", "template": "Summarize: {text}", "description": "A summarizer"}, + ) + assert resp.status_code == 201 + body = resp.json() + assert body["name"] == "summarize" + assert body["template"] == "Summarize: {text}" + assert body["description"] == "A summarizer" + assert "id" in body + + def test_create_prompt_without_description(self, client: TestClient) -> None: + resp = client.post( + "/items/prompts", + json={"name": "minimal", "template": "Hello"}, + ) + assert resp.status_code == 201 + assert resp.json()["description"] is None + + def test_get_prompt(self, client: TestClient) -> None: + created = client.post( + "/items/prompts", + json={"name": "fetch", "template": "Fetch {url}"}, + ).json() + resp = client.get(f"/items/prompts/{created['id']}") + assert resp.status_code == 200 + assert resp.json()["name"] == "fetch" + + def test_get_missing_prompt_is_404(self, client: TestClient) -> None: + resp = client.get("/items/prompts/nonexistent") + assert resp.status_code == 404 + + def test_update_prompt(self, client: TestClient) -> None: + created = client.post( + "/items/prompts", + json={"name": "old", "template": "Old template"}, + ).json() + resp = client.patch( + f"/items/prompts/{created['id']}", + json={"name": "new", "template": "New template", "description": "Updated"}, + ) + assert resp.status_code == 200 + assert resp.json()["name"] == "new" + assert resp.json()["template"] == "New template" + assert resp.json()["description"] == "Updated" + + def test_partial_update_prompt(self, client: TestClient) -> None: + created = client.post( + "/items/prompts", + json={"name": "keep", "template": "Keep this", "description": "Original"}, + ).json() + resp = client.patch( + f"/items/prompts/{created['id']}", + json={"template": "Changed"}, + ) + assert resp.status_code == 200 + assert resp.json()["name"] == "keep" + assert resp.json()["template"] == "Changed" + assert resp.json()["description"] == "Original" + + def test_delete_prompt(self, client: TestClient) -> None: + created = client.post( + "/items/prompts", + json={"name": "gone", "template": "Bye"}, + ).json() + resp = client.delete(f"/items/prompts/{created['id']}") + assert resp.status_code == 204 + assert client.get(f"/items/prompts/{created['id']}").status_code == 404 + + def test_delete_missing_prompt_is_404(self, client: TestClient) -> None: + resp = client.delete("/items/prompts/nonexistent") + assert resp.status_code == 404 + + def test_create_rejects_duplicate_name(self, client: TestClient) -> None: + client.post("/items/prompts", json={"name": "dup", "template": "A"}) + resp = client.post("/items/prompts", json={"name": "dup", "template": "B"}) + assert resp.status_code == 409 + + +# ── Local Resources ───────────────────────────────────────────────────── + + +class TestLocalResources: + def test_list_empty_by_default(self, client: TestClient) -> None: + assert client.get("/items/resources").json() == [] + + def test_create_resource(self, client: TestClient) -> None: + resp = client.post( + "/items/resources", + json={ + "uri": "file:///docs/readme.md", + "name": "Readme", + "mime_type": "text/markdown", + "content": "# Hello", + }, + ) + assert resp.status_code == 201 + body = resp.json() + assert body["uri"] == "file:///docs/readme.md" + assert body["name"] == "Readme" + assert body["mime_type"] == "text/markdown" + assert body["content"] == "# Hello" + assert "id" in body + + def test_create_resource_minimal(self, client: TestClient) -> None: + resp = client.post( + "/items/resources", + json={"uri": "https://example.com/data", "name": "Data"}, + ) + assert resp.status_code == 201 + assert resp.json()["mime_type"] is None + assert resp.json()["content"] is None + + def test_get_resource(self, client: TestClient) -> None: + created = client.post( + "/items/resources", + json={"uri": "file:///a.txt", "name": "A", "content": "aaa"}, + ).json() + resp = client.get(f"/items/resources/{created['id']}") + assert resp.status_code == 200 + assert resp.json()["content"] == "aaa" + + def test_get_missing_resource_is_404(self, client: TestClient) -> None: + resp = client.get("/items/resources/nonexistent") + assert resp.status_code == 404 + + def test_update_resource(self, client: TestClient) -> None: + created = client.post( + "/items/resources", + json={"uri": "file:///old", "name": "Old", "content": "old"}, + ).json() + resp = client.patch( + f"/items/resources/{created['id']}", + json={"uri": "file:///new", "name": "New", "content": "new"}, + ) + assert resp.status_code == 200 + assert resp.json()["uri"] == "file:///new" + assert resp.json()["content"] == "new" + + def test_partial_update_resource(self, client: TestClient) -> None: + created = client.post( + "/items/resources", + json={"uri": "file:///keep", "name": "Keep", "content": "original"}, + ).json() + resp = client.patch( + f"/items/resources/{created['id']}", + json={"content": "updated"}, + ) + assert resp.status_code == 200 + assert resp.json()["uri"] == "file:///keep" + assert resp.json()["content"] == "updated" + + def test_delete_resource(self, client: TestClient) -> None: + created = client.post( + "/items/resources", + json={"uri": "file:///gone", "name": "Gone"}, + ).json() + resp = client.delete(f"/items/resources/{created['id']}") + assert resp.status_code == 204 + assert client.get(f"/items/resources/{created['id']}").status_code == 404 + + def test_delete_missing_resource_is_404(self, client: TestClient) -> None: + resp = client.delete("/items/resources/nonexistent") + assert resp.status_code == 404 + + def test_create_rejects_duplicate_uri(self, client: TestClient) -> None: + client.post("/items/resources", json={"uri": "file:///dup", "name": "A"}) + resp = client.post("/items/resources", json={"uri": "file:///dup", "name": "B"}) + assert resp.status_code == 409 diff --git a/services/instrumenta/test_stdio.py b/services/instrumenta/test_stdio.py new file mode 100644 index 0000000..4e5949e --- /dev/null +++ b/services/instrumenta/test_stdio.py @@ -0,0 +1,86 @@ +"""Tests for the stdio supervisor and PATH probe. + +The stdio supervisor spawns child MCP servers as subprocesses, connects +via stdin/stdout, and autorestarts with capped exponential backoff on +failure. The PATH probe reports which runtimes are available. +""" + +from __future__ import annotations + +import asyncio +import shutil + +import pytest + +from instrumenta.path_probe import probe_runtimes +from instrumenta.stdio_supervisor import StdioSupervisor + + +class TestPathProbe: + def test_probe_returns_dict(self) -> None: + result = probe_runtimes() + assert isinstance(result, dict) + assert "python3" in result + + def test_python3_found_on_path(self) -> None: + result = probe_runtimes() + assert result["python3"] is True + + def test_nonexistent_binary_not_found(self) -> None: + result = probe_runtimes(runtimes=("nonexistent_binary_xyz",)) + assert result.get("nonexistent_binary_xyz") is False + + def test_reflects_actual_PATH(self) -> None: + result = probe_runtimes() + # If python3 is on PATH (which it must be to run this test), + # it should be found. + assert shutil.which("python3") is not None + assert result["python3"] is True + + +class TestStdioSupervisor: + def test_spawn_and_list_tools(self, tmp_path) -> None: + """Spawn a tiny echo MCP server script and list its tools.""" + script = tmp_path / "echo_server.py" + script.write_text( + 'import sys, json\n' + 'line = sys.stdin.readline()\n' + 'resp = {"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-06-18","capabilities":{},"serverInfo":{"name":"echo","version":"0.1.0"}}}\n' + 'sys.stdout.write(json.dumps(resp) + "\\n")\n' + 'sys.stdout.flush()\n' + 'line = sys.stdin.readline()\n' + 'tools = [{"name":"echo","description":"Echo input","inputSchema":{"type":"object","properties":{"text":{"type":"string"}}}}]\n' + 'resp = {"jsonrpc":"2.0","id":2,"result":{"tools":tools}}\n' + 'sys.stdout.write(json.dumps(resp) + "\\n")\n' + 'sys.stdout.flush()\n' + 'sys.stdin.readline()\n' + ) + supervisor = StdioSupervisor() + try: + tools = asyncio.run( + supervisor.spawn_and_list_tools( + command=f"{sys.executable} {script}", + server_name="echo-test", + ) + ) + assert len(tools) == 1 + assert tools[0]["name"] == "echo-test.echo" + finally: + supervisor.close_all() + + def test_nonexistent_command_raises(self) -> None: + supervisor = StdioSupervisor() + try: + with pytest.raises(Exception): + asyncio.run( + supervisor.spawn_and_list_tools( + command="nonexistent_binary_xyz_12345", + server_name="bad", + ) + ) + finally: + supervisor.close_all() + + +# Need sys for executable path +import sys