diff --git a/docs/mcp-server.md b/docs/mcp-server.md index 99be60e..b307a17 100644 --- a/docs/mcp-server.md +++ b/docs/mcp-server.md @@ -7,9 +7,17 @@ server that lets MCP-aware clients drive bcli. The intended caller is Claude Desktop, but it also works with the official [MCP Inspector](https://github.com/modelcontextprotocol/inspector) and any other client that speaks the spec. -The server is deliberately small (4 read-only tools) and delegates every call -to the bcli CLI as a subprocess. Profile resolution, auth, retry, telemetry, -and the read-only `disable_writes` gate are inherited from the CLI for free. +The server generates its tool list **dynamically** by subprocessing +`bcli describe --format json` once on startup, and delegates every call to the +bcli CLI as a subprocess. Profile resolution, auth, retry, telemetry, and the +read-only `disable_writes` gate are inherited from the CLI for free. New CLI +commands light up automatically as MCP tools; deprecated ones disappear. + +Read commands return parsed stdout JSON. Mutating commands (`bcli_post`, +`bcli_patch`, `bcli_delete`, `bcli_attach_upload`, `bcli_batch_run`) pass +`--result-out ` and return the AIP §Phase 2 result envelope content as +the tool result. A `status="failed"` envelope surfaces as an MCP `ToolError` +with the BC correlation id quoted so the agent can cite it. ## Install @@ -47,8 +55,15 @@ Add a `mcpServers` entry to `~/Library/Application Support/Claude/claude_desktop } ``` -Restart Claude Desktop. You should see four tools register: `query`, -`list_endpoints`, `describe_endpoint`, `list_companies`. +Restart Claude Desktop. The server registers one tool per command in +`bcli describe`'s output — typically ~20 tools, one per read/mutating verb in +the CLI. Use the MCP Inspector to enumerate them, or run +`bcli describe --format json` directly to preview what'll appear. + +Renames from the pre-Phase-5 surface: the old hand-written `query`, +`list_endpoints`, `describe_endpoint`, and `list_companies` are now +`bcli_get`, `bcli_endpoint_list`, `bcli_endpoint_info`, and `bcli_company_list` +respectively (matching the CLI subcommand paths). If `bcli-mcp` isn't on Claude Desktop's PATH (uv tool install paths can be tricky), use the full path: @@ -66,17 +81,25 @@ tricky), use the full path: ## Tool surface +The tool list is generated from `bcli describe --format json` on startup, so +the canonical reference is the describe output for your install. The most +useful subset: + | Tool | What it does | Notes | |------|--------------|-------| -| `query` | Run an OData query against an entity. | `top` defaults to 50, capped at 1000. Use `select` to keep payloads small. | -| `list_endpoints` | List entities the active profile can reach. | Honours `disable_standard_api`, `allowed_categories`, `allowed_endpoints`. | -| `describe_endpoint` | Show fields, key, supported ops, and route for one entity. | `fields` is populated only after `bcli endpoint fields ` has been run. | -| `list_companies` | Companies on the active environment. | Returns `[{id, name, alias, is_default}]`. | - -Mutating commands (`post` / `patch` / `delete`), file uploads, batch runs, and -admin/setup flows are deliberately not exposed. Claude can always fall back to -`Bash` + `bcli` directly when those are needed — and that path trips the -existing `disable_writes` confirmation prompt. +| `bcli_get` | Run an OData query against an entity. | `top` defaults to 50, capped at 1000 (carried from describe's `limits`). Use `select` to keep payloads small. | +| `bcli_endpoint_list` | List entities the active profile can reach. | Honours `disable_standard_api`, `allowed_categories`, `allowed_endpoints`. | +| `bcli_endpoint_info` | Show fields, key, supported ops, and route for one entity. | `fields` is populated only after `bcli endpoint fields ` has been run. | +| `bcli_endpoint_fields` | Discover the fields for one entity and persist them to the local registry. | One BC API call; populates the cache for every future call. | +| `bcli_company_list` | Companies on the active environment. | Returns `[{id, name, alias, is_default}]`. | +| `bcli_q` | Run a saved query by name with `${{ params.X }}` substitution. | The "daily questions" surface — hides OData syntax. | +| `bcli_post` / `bcli_patch` / `bcli_delete` | Mutating verbs. | Server passes `--result-out ` and returns the AIP §Phase 2 result envelope as the tool result. `status="failed"` raises `ToolError` with the BC correlation id. | +| `bcli_attach_upload` | Two-phase document attachment upload. | Same envelope contract as the other mutating verbs. | +| `bcli_batch_run` | Execute a YAML batch file. | Returns an envelope whose `record_id` is the batch ledger run id; pivot to `bcli_describe`'s batch state subcommand for per-step detail. | +| `bcli_describe` | Re-emit the full describe payload. | The MCP server itself uses this on startup; tools can call it to discover what else is exposed. | + +`auth login`, `config init`, and other interactive commands (`effects: +["other"]` in describe) are filtered out — those are command-line-only. ## Trust model — why the server resets cwd @@ -95,20 +118,20 @@ sources are then exactly: Per-tool calls do not honour a per-request `cwd` argument. The server runs with a single fixed working directory for its lifetime. -## BC query objects vs entity pages — what to expect from `query()` +## BC query objects vs entity pages — what to expect from `bcli_get` Not every endpoint in BC's OData surface is a fully-featured entity. Some are "query objects" — read-only summary pages exposed via OData (e.g. `customerSales`, `vendorPurchases`). They behave like entities for `GET` but Microsoft's runtime drops `$orderby` and `$filter` support on most of -them. A `query()` call against one of these with `orderby=` or `filter=` +them. A `bcli_get` call against one of these with `orderby=` or `filter=` will 400 from BC. How to recognise one: there's no flag in the registry today (it'd require a hint per endpoint), so the practical signal is the 400 itself. The recovery pattern that works: -1. `query(entity="customerSales", top=1000)` — pull a bounded page. +1. `bcli_get(endpoint="customerSales", top=1000)` — pull a bounded page. 2. Sort the result client-side in your reasoning step. 3. Take the top N. @@ -119,24 +142,22 @@ you may miss the actual top customer. For BC tenants with very large summary pages, fall back to `bcli get …` via Bash with `--all` (which follows pagination). -Entity pages (most of `bcli endpoint list`) support full OData. Use -`describe_endpoint(name)` to see whether `fields_discovered` is `true` +Entity pages (most of `bcli_endpoint_list`) support full OData. Use +`bcli_endpoint_info(name)` to see whether `fields_discovered` is `true` and what `fields` look like; the registry doesn't currently track which endpoints are query objects vs entities, so you'll learn this empirically. -## Discovering field names — `discover_fields` +## Discovering field names -`describe_endpoint(name)` returns `fields: []` and `fields_discovered: +`bcli_endpoint_info(name)` returns `fields: []` and `fields_discovered: false` if the local registry hasn't probed BC for that entity yet. Two ways to recover, in order of cost: -1. Cheapest: call `query(entity=name, top=1)` once and read the keys off - the returned record. No registry mutation, zero cache pollution. -2. One-time: pass `discover_fields=true` to `describe_endpoint`. The - tool runs `bcli endpoint fields ` first, which fetches one - record, persists the field names to the local registry, and then - returns the populated metadata. Every subsequent call (any user, any - session) gets `fields_discovered: true` for free. +1. Cheapest: call `bcli_get(endpoint=name, top=1)` once and read the keys + off the returned record. No registry mutation, zero cache pollution. +2. One-time: call `bcli_endpoint_fields(name)`. That fetches one record, + persists the field names to the local registry, and every subsequent + call (any user, any session) gets `fields_discovered: true` for free. Pick (1) for one-shot analysis. Pick (2) when the entity is one you'll revisit a lot — the registry-cached field list also feeds bcli's @@ -148,9 +169,11 @@ Pairing an MCP server with a CLI tool is empirically token-favorable for **bounded, schema-stable** responses. The OSS server ships with two guard-rails baked in: -* `query.top` defaults to 50 (max 1000) — an unbounded request can't pull a - whole table into context. -* Tool docstrings are short. The schema-payload Claude sees is small. +* `bcli_get.top` defaults to 50 (max 1000) — the safety bound is carried from + `bcli describe`'s `limits` field, so an unbounded request can't pull a whole + table into context. +* Tool descriptions come straight from the CLI's docstrings. They're short by + construction — the schema-payload Claude sees is small. It is **not** universally a token win. For browse-style "show me everything" workflows, falling back to `bcli get --format markdown` via Bash is diff --git a/src/bcli/result_envelope.py b/src/bcli/result_envelope.py index 07d5bfe..5fdbf8a 100644 --- a/src/bcli/result_envelope.py +++ b/src/bcli/result_envelope.py @@ -107,4 +107,34 @@ def write_envelope( os.close(fd) -__all__ = ["ENVELOPE_VERSION", "ResultEnvelope", "write_envelope"] +def read_envelope(path: Path) -> ResultEnvelope: + """Inverse of :func:`write_envelope` — load a JSON envelope file. + + Used by ``bcli_mcp`` to pick up the result of a mutating CLI + invocation. Tolerates missing optional fields so an older envelope + written by a previous bcli version still loads (forward-compat). + """ + raw = json.loads(Path(path).read_text(encoding="utf-8")) + return ResultEnvelope( + version=raw.get("version", ENVELOPE_VERSION), + invocation_id=raw["invocation_id"], + tool_version=raw.get("tool_version", ""), + profile=raw.get("profile"), + environment=raw.get("environment"), + company=raw.get("company"), + method=raw["method"], + endpoint=raw["endpoint"], + resolved_url=raw.get("resolved_url"), + record_id=raw.get("record_id"), + dry_run=bool(raw.get("dry_run", False)), + status=raw["status"], + exit_code=int(raw["exit_code"]), + bc_correlation_id=raw.get("bc_correlation_id"), + telemetry_event_id=raw.get("telemetry_event_id"), + audit_log_offset=raw.get("audit_log_offset"), + started_at=raw.get("started_at", ""), + duration_ms=int(raw.get("duration_ms", 0)), + ) + + +__all__ = ["ENVELOPE_VERSION", "ResultEnvelope", "read_envelope", "write_envelope"] diff --git a/src/bcli_cli/commands/describe_cmd.py b/src/bcli_cli/commands/describe_cmd.py index 76e5b33..8411043 100644 --- a/src/bcli_cli/commands/describe_cmd.py +++ b/src/bcli_cli/commands/describe_cmd.py @@ -160,30 +160,86 @@ def _supported_formats_from_signature(sig: inspect.Signature) -> list[str]: return ["json"] -def _options_from_signature(sig: inspect.Signature) -> list[dict[str, Any]]: +# Safety bounds the MCP server's auto-generated tools must honour. Keys +# are the (command-path-tuple, long-flag-name); the value rides as the +# option's ``limits`` sub-object in describe's JSON output. This is the +# *one* place that says "an agent must clamp this before invoking" — +# Phase 5's tool generator (``bcli_mcp._tool_generator``) reads it +# straight off describe. +_OPTION_LIMITS: dict[tuple[tuple[str, ...], str], dict[str, Any]] = { + (("get",), "--top"): {"default": 50, "minimum": 1, "maximum": 1000}, +} + + +def _options_from_signature( + sig: inspect.Signature, *, path: tuple[str, ...] = (), +) -> list[dict[str, Any]]: options: list[dict[str, Any]] = [] for name, param in sig.parameters.items(): info = param.default param_decls = getattr(info, "param_decls", None) if not param_decls: - # Positional argument — describe lists options only. + # Positional argument — handled by ``_positionals_from_signature``. continue # Pick the longest decl (typically the ``--foo`` form). long_name = sorted(param_decls, key=lambda d: (-len(d), d))[0] type_name = _annotation_to_name(param.annotation) opt: dict[str, Any] = {"name": long_name, "type": type_name} + # Required options expose ``typer.Option(..., ...)`` — the + # default is the literal ellipsis. Phase 5's tool generator + # marks these as required in the JSON Schema so an agent + # doesn't construct a missing-argument call. + option_default = getattr(info, "default", None) + if option_default is Ellipsis: + opt["required"] = True # Crude validator hint — the spec example shows # ``validates: "odata-filter"`` for ``--filter``. if long_name == "--filter": opt["validates"] = "odata-filter" + # Optional safety-bounds for clamp-on-call (Phase 5 MCP wiring). + limits = _OPTION_LIMITS.get((path, long_name)) + if limits is not None: + opt["limits"] = dict(limits) options.append(opt) return options +def _positionals_from_signature(sig: inspect.Signature) -> list[dict[str, Any]]: + """List of positional arguments (Typer ``Argument``) per command. + + Each entry has ``{name, type, required}``. ``required=True`` when + the parameter has no default; ``False`` when it does (Typer renders + optional positionals via ``typer.Argument(None, ...)``). + """ + positionals: list[dict[str, Any]] = [] + for name, param in sig.parameters.items(): + info = param.default + param_decls = getattr(info, "param_decls", None) + if param_decls: + # Option / flag — skip. + continue + # ``typer.Argument(...)`` instances expose ``default`` on the info + # object; a missing default (``...``) means required. + argument_default = getattr(info, "default", None) + required = argument_default is Ellipsis or argument_default is None + # ``typer.Argument(None, ...)`` is the idiomatic optional form + # — default is ``None`` and the function accepts it. Treat as + # not-required to match the CLI's behaviour. + if argument_default is None: + required = False + positionals.append({ + "name": name, + "type": _annotation_to_name(param.annotation), + "required": bool(required), + }) + return positionals + + def _annotation_to_name(ann: Any) -> str: """Render a type annotation as a JSON-friendly string. - ``Optional[int]`` → ``"int"``, ``bool`` → ``"bool"``, etc. + ``Optional[int]`` → ``"int"``, ``bool`` → ``"bool"``, + ``list[str]`` → ``"list[str]"`` (preserved so MCP can branch on it). """ if ann is inspect.Parameter.empty: return "string" @@ -192,7 +248,15 @@ def _annotation_to_name(ann: Any) -> str: s = ann else: s = getattr(ann, "__name__", None) or str(ann) - s = s.replace("Optional[", "").rstrip("]") + s = s.replace("Optional[", "") + # Strip one trailing ``]`` that the Optional unwrap leaves behind. + if s.endswith("]") and s.count("[") < s.count("]"): + s = s[:-1] + # Take the last dotted component but preserve the bracket payload. + if "[" in s: + head, rest = s.split("[", 1) + head = head.split(".")[-1].lower() + return f"{head}[{rest}".rstrip() s = s.split(".")[-1] return s.lower() or "string" @@ -223,7 +287,8 @@ def _walk_typer(typer_app, parent_path: tuple[str, ...] = ()) -> list[dict[str, entry: dict[str, Any] = { "path": list(path), "summary": _summary_from_callback(callback), - "options": _options_from_signature(sig), + "options": _options_from_signature(sig, path=path), + "positionals": _positionals_from_signature(sig), "effects": _classify_effects(path), "supported_formats": _supported_formats_from_signature(sig), } diff --git a/src/bcli_mcp/__main__.py b/src/bcli_mcp/__main__.py index 8f50786..f9d585b 100644 --- a/src/bcli_mcp/__main__.py +++ b/src/bcli_mcp/__main__.py @@ -22,7 +22,7 @@ def main() -> None: os.chdir(os.path.expanduser("~")) try: - from bcli_mcp._server import mcp + from bcli_mcp._server import get_server except ImportError as exc: sys.stderr.write( f"bcli-mcp: failed to import server: {exc}\n" @@ -30,7 +30,7 @@ def main() -> None: ) raise SystemExit(1) from exc - mcp.run() + get_server().run() if __name__ == "__main__": diff --git a/src/bcli_mcp/_runner.py b/src/bcli_mcp/_runner.py index a2cac60..e044892 100644 --- a/src/bcli_mcp/_runner.py +++ b/src/bcli_mcp/_runner.py @@ -6,6 +6,16 @@ JSON response, and surface a clean ``ToolError`` on non-zero exits with Rich markup stripped from the error message. +Two entry points: + +* :func:`run_bcli_json` — read-only invocations. Append ``--format json``, + parse stdout, return the parsed value. +* :func:`run_bcli_with_envelope` — mutating invocations. Append + ``--result-out `` and ``--format json``; return + ``(exit_code, stdout, stderr)`` so the caller reads the envelope file + back via :func:`bcli.result_envelope.read_envelope`. The envelope is + the source of truth for outcome; stdout is mostly noise. + We deliberately do NOT import ``bcli`` Python modules here. Subprocess delegation is the design — see ``docs/mcp-server.md``. """ @@ -16,7 +26,7 @@ import os import re import subprocess -from typing import Any +from typing import Any, Mapping # Imported lazily so the runner module is importable in environments # that don't have the optional ``mcp`` package installed (tests). @@ -37,6 +47,13 @@ def _strip_rich(text: str) -> str: return _RICH_MARKUP.sub("", text).strip() +def _env_with_profile(profile: str | None) -> dict[str, str]: + env = os.environ.copy() + if profile: + env["BCLI_PROFILE"] = profile + return env + + def run_bcli_json( *args: str, profile: str | None = None, @@ -57,17 +74,13 @@ def run_bcli_json( argv.extend(args) argv.extend(["--format", "json"]) - env = os.environ.copy() - if profile: - env["BCLI_PROFILE"] = profile - try: proc = subprocess.run( argv, capture_output=True, text=True, timeout=timeout, - env=env, + env=_env_with_profile(profile), check=False, ) except FileNotFoundError as exc: @@ -99,36 +112,33 @@ def run_bcli_json( ) from exc -def run_bcli_side_effect( - *args: str, - profile: str | None = None, - timeout: float = 120.0, -) -> None: - """Run ``bcli `` for its side effect; ignore stdout content. +def run_bcli_with_envelope( + args: list[str], + *, + env: Mapping[str, str] | None, + capture_envelope_path: str, + timeout: float = 300.0, +) -> tuple[int, str, str]: + """Run a mutating ``bcli`` command with ``--result-out ``. - Some bcli subcommands (e.g. ``endpoint fields``) emit human-readable - text on stdout but persist their useful work to the local registry as - a side effect. We don't want that text — we want the cache write. + Returns ``(exit_code, stdout, stderr)``. The caller is expected to + read the envelope file at ``capture_envelope_path`` via + :func:`bcli.result_envelope.read_envelope` — that's the source of + truth for the mutation outcome. - Raises ``ToolError`` on non-zero exit (with Rich markup stripped) or - timeout. Otherwise silent on success. + The exit code is passed through so the caller can sanity-check (an + envelope missing on disk + a non-zero exit code is "bcli crashed + before writing the envelope" rather than "the mutation succeeded"). """ - argv = ["bcli"] - if profile: - argv.extend(["--profile", profile]) - argv.extend(args) - - env = os.environ.copy() - if profile: - env["BCLI_PROFILE"] = profile - + argv = ["bcli", *args, "--result-out", capture_envelope_path, "--format", "json"] + effective_env = dict(env) if env is not None else os.environ.copy() try: proc = subprocess.run( argv, capture_output=True, text=True, timeout=timeout, - env=env, + env=effective_env, check=False, ) except FileNotFoundError as exc: @@ -141,8 +151,4 @@ def run_bcli_side_effect( f"bcli {' '.join(args)} timed out after {timeout}s" ) from exc - if proc.returncode != 0: - message = _strip_rich(proc.stderr or proc.stdout or "(no output)") - raise _ToolError( - f"bcli {' '.join(args)} exited {proc.returncode}: {message}" - ) + return proc.returncode, proc.stdout, proc.stderr diff --git a/src/bcli_mcp/_server.py b/src/bcli_mcp/_server.py index 4daa927..78ded55 100644 --- a/src/bcli_mcp/_server.py +++ b/src/bcli_mcp/_server.py @@ -1,162 +1,243 @@ -"""FastMCP server with the day-1 bcli tool surface (4 tools). - -Design constraints: - -* Token economy. Tool docstrings are short and concrete. ``query`` caps - ``top`` at 1000 with a sane default of 50 so an unbounded request can't - pull a whole table into context. -* Subprocess only. Every tool delegates to ``run_bcli_json`` so profile - resolution, auth, retry, telemetry, and the registry filters that - ``bcli_cli._state`` applies (``disable_standard_api``, - ``allowed_categories``) are inherited for free. -* Read-only. Mutating commands (post/patch/delete, batch, attach upload) - are deliberately NOT exposed — they go through the CLI directly where - the existing ``disable_writes`` prompt protects. +"""FastMCP server with a dynamically-generated bcli tool surface. + +Phase 5 of AIP v0.1. The server's tool list is no longer hand-written. +On startup, ``_build_server`` subprocesses ``bcli describe --format json`` +and turns each command into one FastMCP tool via +:func:`bcli_mcp._tool_generator.generate_tools`. New CLI commands light +up automatically; deprecated ones disappear. + +Two invocation paths: + +* **Read** (``effects: ["read"]``) — append ``--format json``, parse + stdout, return the parsed JSON. +* **Mutating** (``emits_result_envelope: True``) — pass + ``--result-out ``, read the envelope back, return its content + as the tool result. A ``status="failed"`` envelope surfaces as an + MCP ``ToolError`` with the envelope's correlation id quoted for the + agent to cite. + +If ``bcli describe`` fails on startup (bcli not on PATH, broken install, +etc.), the server still starts — with zero tools registered and a stderr +warning. The operator can fix the install and reconnect. + +The subprocess boundary inherits auth, retry, telemetry, profile gates +for free — that's the whole point. No Python imports from ``bcli`` core. """ from __future__ import annotations +import inspect +import json +import os +import subprocess +import sys +import tempfile from typing import Any from mcp.server.fastmcp import FastMCP - -from bcli_mcp._runner import run_bcli_json, run_bcli_side_effect - -mcp = FastMCP("bcli") - -# Hard caps so an agent can't accidentally pull a whole table. -_QUERY_DEFAULT_TOP = 50 -_QUERY_MAX_TOP = 1000 - - -@mcp.tool() -async def query( - entity: str, - filter: str | None = None, - select: str | None = None, - top: int | None = None, - skip: int | None = None, - orderby: str | None = None, - expand: str | None = None, - record_id: str | None = None, - publisher: str | None = None, - group: str | None = None, - version: str | None = None, - profile: str | None = None, -) -> list[dict[str, Any]]: - """Run an OData query against a Business Central entity. - - Returns a list of records. ``top`` defaults to 50 and is capped at - 1000 — for browse-style "show me everything" use the bcli CLI directly. - Use ``select`` to limit fields and keep responses small. - - ``entity`` is the OData entity-set name (e.g. ``customers``, - ``salesInvoices``). Use ``list_endpoints`` to discover what's - available; ``describe_endpoint`` to see the field shape. - """ - effective_top = top if top is not None else _QUERY_DEFAULT_TOP - if effective_top < 1: - effective_top = 1 - if effective_top > _QUERY_MAX_TOP: - effective_top = _QUERY_MAX_TOP - - args: list[str] = ["get", entity] - if record_id: - args.append(record_id) - if filter: - args.extend(["--filter", filter]) - if select: - args.extend(["--select", select]) - args.extend(["--top", str(effective_top)]) - if skip is not None: - args.extend(["--skip", str(skip)]) - if orderby: - args.extend(["--orderby", orderby]) - if expand: - args.extend(["--expand", expand]) - if publisher: - args.extend(["--publisher", publisher]) - if group: - args.extend(["--group", group]) - if version: - args.extend(["--version", version]) - - result = run_bcli_json(*args, profile=profile) - # `bcli get ` returns a list. `bcli get ` - # returns a single object — wrap so the tool's return type is stable. - if isinstance(result, dict): - return [result] - return result - - -@mcp.tool() -async def list_endpoints( - category: str | None = None, - custom_only: bool = False, - standard_only: bool = False, - profile: str | None = None, -) -> list[dict[str, Any]]: - """List Business Central entities the active profile can reach. - - Returns ``[{name, category, custom, supported_ops, key_field, - publisher, group, version, description}]``. Honours profile-level - filters (``disable_standard_api``, ``allowed_categories``). +from mcp.server.fastmcp.exceptions import ToolError + +from bcli.result_envelope import read_envelope +from bcli_mcp._runner import ( + _strip_rich, + run_bcli_json, + run_bcli_with_envelope, +) +from bcli_mcp._tool_generator import GeneratedTool, generate_tools + + +# Map JSON Schema primitive types back to Python types so FastMCP's +# pydantic-based signature introspection picks up the right types. +_PYTHON_TYPE_FOR_JSON: dict[str, type] = { + "string": str, + "integer": int, + "number": float, + "boolean": bool, +} + + +def _python_type(json_type: str) -> type: + return _PYTHON_TYPE_FOR_JSON.get(json_type, str) + + +def _apply_dynamic_signature(handler, tool: GeneratedTool) -> None: + """Override ``handler`` so its signature matches the tool's input + schema. FastMCP's tool registration walks the function signature + (via pydantic) to build the JSON Schema agents see — without this + override every tool would show up as ``inputs: object`` and lose + its named parameters.""" + properties = tool.input_schema.get("properties", {}) + required = set(tool.input_schema.get("required", [])) + + params: list[inspect.Parameter] = [] + annotations: dict[str, Any] = {} + for name, prop in properties.items(): + py_type = _python_type(prop.get("type", "string")) + default = prop.get("default") if name not in required else inspect.Parameter.empty + param = inspect.Parameter( + name=name, + kind=inspect.Parameter.KEYWORD_ONLY, + default=inspect.Parameter.empty if name in required else default, + annotation=py_type, + ) + params.append(param) + annotations[name] = py_type + # Every tool also accepts an optional ``profile`` override so the + # agent can scope a single call without restarting the server. + if "profile" not in annotations: + params.append(inspect.Parameter( + name="profile", + kind=inspect.Parameter.KEYWORD_ONLY, + default=None, + annotation=str | None, + )) + annotations["profile"] = str | None + annotations["return"] = Any + handler.__signature__ = inspect.Signature(params) + handler.__annotations__ = annotations + + +def _load_describe_payload() -> dict[str, Any]: + """Subprocess ``bcli describe --format json`` and return parsed JSON. + + Raises on any failure — caller decides whether to fall back to a + zero-tools server. """ - args = ["endpoint", "list"] - if custom_only: - args.append("--custom") - if standard_only: - args.append("--standard") - if category: - args.extend(["--category", category]) - return run_bcli_json(*args, profile=profile) - - -@mcp.tool() -async def describe_endpoint( - name: str, - discover_fields: bool = False, - profile: str | None = None, -) -> dict[str, Any]: - """Show fields, key, supported operations, and route for one entity. - - The response includes ``fields_discovered: bool``. When false, ``fields`` - is empty *because the local registry hasn't probed BC for this entity - yet*, not because the entity has no fields. Two ways to recover: - - * Pass ``discover_fields=True`` (this tool runs ``bcli endpoint fields - `` first — costs one BC API call, populates the cache for every - future call). - * Or call ``query(entity=name, top=1)`` and read the keys off the first - record. Cheaper if you only need the schema for one analysis. - - Note: some BC pages are "query objects" (read-only summary views) which - don't support server-side ``$orderby`` or ``$filter``. If a ``query()`` - call against this entity 400s on those, fall back to client-side sort - over a bounded ``top`` window — see docs/mcp-server.md. + proc = subprocess.run( + ["bcli", "describe", "--format", "json"], + capture_output=True, text=True, timeout=30.0, check=False, + ) + if proc.returncode != 0: + raise RuntimeError( + f"bcli describe exited {proc.returncode}: " + f"{_strip_rich(proc.stderr or proc.stdout or '(no output)')}" + ) + return json.loads(proc.stdout) + + +def _make_read_handler(tool: GeneratedTool): + """Return an async callable FastMCP will register as the tool fn. + + The signature is rewritten by :func:`_apply_dynamic_signature` so + FastMCP's pydantic-based introspection sees one keyword arg per + input-schema property — preserves the tool's input schema as the + agent actually receives it. """ - if discover_fields: - # ``bcli endpoint fields`` fetches one record from BC and persists - # the field names into the local registry. The text output is - # discarded — we only care about the cache write. If discovery - # fails (e.g. entity needs a filter, no records returned), the - # subsequent ``endpoint info`` call returns ``fields_discovered: - # false`` and the agent can fall back to a probe query. + async def handler(**inputs: Any) -> Any: + # Drop None values so optional args that the agent didn't fill + # in don't pollute the argv builder. + inputs = {k: v for k, v in inputs.items() if v is not None} + profile = inputs.pop("profile", None) + args = tool.build_argv(inputs) + result = run_bcli_json(*args, profile=profile) + # Single-record dict from ``bcli get `` is wrapped + # so the return type stays stable for the agent. + if tool.name == "bcli_get" and isinstance(result, dict): + return [result] + return result + handler.__name__ = tool.name + handler.__doc__ = tool.description + _apply_dynamic_signature(handler, tool) + return handler + + +def _make_mutating_handler(tool: GeneratedTool): + """Mutating handler: passes ``--result-out``, reads envelope back.""" + async def handler(**inputs: Any) -> dict[str, Any]: + inputs = {k: v for k, v in inputs.items() if v is not None} + profile = inputs.pop("profile", None) + args = list(tool.build_argv(inputs)) + if profile: + args = ["--profile", profile, *args] + + env = os.environ.copy() + if profile: + env["BCLI_PROFILE"] = profile + + # The envelope is written atomically by bcli; we just need a + # private path the subprocess can write to and we can read. + fd, env_path = tempfile.mkstemp(prefix="bcli-mcp-", suffix=".json") + os.close(fd) try: - run_bcli_side_effect("endpoint", "fields", name, profile=profile) - except Exception: - pass - - return run_bcli_json("endpoint", "info", name, profile=profile) + exit_code, _stdout, stderr = run_bcli_with_envelope( + args, env=env, capture_envelope_path=env_path, + ) + try: + envelope = read_envelope(env_path) + except (OSError, ValueError, KeyError) as exc: + # Envelope missing or malformed → bcli crashed before + # writing. Surface the stderr as the error so the agent + # can read it. + raise ToolError( + f"bcli {' '.join(args)} exited {exit_code} without " + f"writing an envelope: {_strip_rich(stderr or '(no output)')}" + ) from exc + + if envelope.status == "failed": + corr = envelope.bc_correlation_id or "n/a" + raise ToolError( + f"bcli {' '.join(args)} failed (exit_code=" + f"{envelope.exit_code}, correlation_id={corr})" + ) + # Return the envelope as a plain dict so MCP can serialize. + from dataclasses import asdict + return asdict(envelope) + finally: + try: + os.unlink(env_path) + except OSError: + pass + + handler.__name__ = tool.name + handler.__doc__ = tool.description + _apply_dynamic_signature(handler, tool) + return handler + + +def _register_tool(mcp: FastMCP, tool: GeneratedTool) -> None: + """Register one generated tool with FastMCP. + + Read handlers get a thin wrapper; mutating handlers get the + envelope-reading flow. Profile is always available as an extra + kwarg the agent can pass to scope the call. + """ + if tool.emits_envelope: + fn = _make_mutating_handler(tool) + else: + fn = _make_read_handler(tool) + mcp.tool(name=tool.name, description=tool.description)(fn) -@mcp.tool() -async def list_companies( - profile: str | None = None, -) -> list[dict[str, Any]]: - """List companies on the active environment. +def _build_server(*, describe_payload: dict[str, Any] | None = None) -> FastMCP: + """Build a FastMCP server with tools generated from describe. - Returns ``[{id, name, alias, is_default}]``. ``alias`` is the local - nickname configured via ``bcli company alias …`` (null if unset). + ``describe_payload`` is injected for tests; production calls + :func:`_load_describe_payload` to subprocess the real CLI. """ - return run_bcli_json("company", "list", profile=profile) + mcp = FastMCP("bcli") + if describe_payload is None: + try: + describe_payload = _load_describe_payload() + except Exception as exc: # noqa: BLE001 + sys.stderr.write( + f"bcli-mcp: warning — could not load describe payload " + f"({exc}). Starting with zero tools; fix the install and " + "reconnect.\n" + ) + return mcp + for tool in generate_tools(describe_payload): + _register_tool(mcp, tool) + return mcp + + +# Module-level singleton used by ``python -m bcli_mcp``. Built lazily so +# the import is cheap and tests can patch ``_build_server`` cleanly. +mcp: FastMCP | None = None + + +def get_server() -> FastMCP: + """Return the module-level server, building it on first call.""" + global mcp + if mcp is None: + mcp = _build_server() + return mcp diff --git a/src/bcli_mcp/_tool_generator.py b/src/bcli_mcp/_tool_generator.py new file mode 100644 index 0000000..974d8ff --- /dev/null +++ b/src/bcli_mcp/_tool_generator.py @@ -0,0 +1,195 @@ +"""Dynamic MCP tool generator — parses ``bcli describe`` output. + +Phase 5 of AIP v0.1. The contract is: any tool surface an MCP client sees +comes directly from what ``bcli describe --format json`` projects. +No hand-written tool schemas allowed. New CLI commands light up +automatically; deprecated ones disappear. + +Each command in describe becomes one :class:`GeneratedTool`: + +* ``name`` — ``bcli_`` +* ``description`` — the command's summary +* ``input_schema`` — JSON Schema built from ``positionals`` + ``options`` + (with ``limits`` carried through for safety bounds) +* ``effects`` — passed through from describe (``["read"]`` / + ``["mutating"]``) +* ``emits_envelope`` — true iff the command declared + ``emits_result_envelope`` (mutating verbs do; read + verbs don't) +* ``build_argv(inputs)`` — turns a tool call's kwargs into the right + bcli CLI argv (positionals first, then ``--flag + value`` pairs; bool flags only when True) +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +# describe ``type`` strings → JSON Schema primitive types. +_TYPE_MAP: dict[str, str] = { + "str": "string", + "string": "string", + "int": "integer", + "integer": "integer", + "bool": "boolean", + "boolean": "boolean", + "float": "number", + "path": "string", +} + + +def _is_list_type(type_name: str) -> bool: + """True if describe described a list-of-strings (or similar) input. + + ``bcli describe`` types these as ``list[str]``. The MCP tool surfaces + them as a single string the agent fills with space-separated tokens + (e.g. ``"batch run"`` for ``bcli describe batch run``). build_argv + splits on whitespace at call time so the bcli subprocess gets the + right argv shape. + """ + return type_name.lower().startswith("list[") + + +def _path_to_tool_name(path: list[str]) -> str: + """``["batch", "list-templates"]`` → ``"bcli_batch_list_templates"``.""" + flat = "_".join(p.replace("-", "_") for p in path) + return f"bcli_{flat}" + + +def _flag_to_kwarg(flag: str) -> str: + """``"--filter"`` → ``"filter"``; ``"--no-registry"`` → ``"no_registry"``.""" + return flag.lstrip("-").replace("-", "_") + + +def _option_to_input_property(opt: dict[str, Any]) -> dict[str, Any]: + """Build the JSON Schema property entry for one ``option`` entry.""" + json_type = _TYPE_MAP.get(opt.get("type", "string"), "string") + prop: dict[str, Any] = {"type": json_type} + limits = opt.get("limits") or {} + if "default" in limits: + prop["default"] = limits["default"] + if "minimum" in limits: + prop["minimum"] = limits["minimum"] + if "maximum" in limits: + prop["maximum"] = limits["maximum"] + return prop + + +@dataclass(frozen=True) +class GeneratedTool: + """One MCP tool, projected from a single describe command entry.""" + + name: str + description: str + path: tuple[str, ...] + positionals: tuple[dict[str, Any], ...] + options: tuple[dict[str, Any], ...] + effects: list[str] + emits_envelope: bool + input_schema: dict[str, Any] = field(default_factory=dict) + + def build_argv(self, inputs: dict[str, Any]) -> list[str]: + """Translate a tool-call ``inputs`` dict into bcli CLI argv. + + Positionals come first in declared order; missing required + positionals raise ``ValueError`` so the caller never ships + malformed argv. Boolean flags emit only when truthy. Other + flags emit ``--name `` when the input is not ``None``. + """ + argv: list[str] = list(self.path) + + # Positionals. + for pos in self.positionals: + name = pos["name"] + value = inputs.get(name) + if value is None: + if pos.get("required"): + raise ValueError( + f"required positional '{name}' missing for {self.name}", + ) + continue + # ``list[str]`` positionals are passed as a single + # whitespace-separated string at the MCP surface and split + # here at argv-build time. Lets ``bcli_describe`` accept + # ``command_path="batch run"`` and produce + # ``bcli describe batch run``. + if _is_list_type(str(pos.get("type", ""))): + if isinstance(value, list): + argv.extend(str(v) for v in value) + else: + argv.extend(str(value).split()) + continue + argv.append(str(value)) + + # Options. + for opt in self.options: + kwarg = _flag_to_kwarg(opt["name"]) + value = inputs.get(kwarg) + if value is None: + continue + json_type = _TYPE_MAP.get(opt.get("type", "string"), "string") + if json_type == "boolean": + if value: + argv.append(opt["name"]) + continue + argv.extend([opt["name"], str(value)]) + return argv + + +def _build_input_schema( + positionals: list[dict[str, Any]], options: list[dict[str, Any]], +) -> dict[str, Any]: + """Compose the MCP tool input schema.""" + properties: dict[str, Any] = {} + required: list[str] = [] + + for pos in positionals: + properties[pos["name"]] = { + "type": _TYPE_MAP.get(pos.get("type", "string"), "string"), + } + if pos.get("required"): + required.append(pos["name"]) + + for opt in options: + kwarg = _flag_to_kwarg(opt["name"]) + properties[kwarg] = _option_to_input_property(opt) + if opt.get("required"): + required.append(kwarg) + + return { + "type": "object", + "properties": properties, + "required": required, + } + + +def generate_tools(payload: dict[str, Any]) -> list[GeneratedTool]: + """Yield one :class:`GeneratedTool` per command entry in ``payload``. + + Commands with ``effects: ["other"]`` are skipped — interactive + commands (``auth login``, ``config init``) aren't scriptable. + """ + out: list[GeneratedTool] = [] + for cmd in payload.get("commands", []): + effects = list(cmd.get("effects", [])) + if effects == ["other"]: + continue + positionals = list(cmd.get("positionals") or []) + options = list(cmd.get("options") or []) + tool = GeneratedTool( + name=_path_to_tool_name(cmd["path"]), + description=cmd.get("summary", "") or "", + path=tuple(cmd["path"]), + positionals=tuple(positionals), + options=tuple(options), + effects=effects, + emits_envelope=bool(cmd.get("emits_result_envelope", False)), + input_schema=_build_input_schema(positionals, options), + ) + out.append(tool) + return out + + +__all__ = ["GeneratedTool", "generate_tools"] diff --git a/tests/test_describe/test_describe_positionals_limits.py b/tests/test_describe/test_describe_positionals_limits.py new file mode 100644 index 0000000..59ba506 --- /dev/null +++ b/tests/test_describe/test_describe_positionals_limits.py @@ -0,0 +1,119 @@ +"""Phase 5 additive describe fields — ``positionals`` and option ``limits``. + +The MCP server (Phase 5) generates its tool list from describe. To do +that faithfully it needs: + +* The positional arguments per command (``bcli post `` — + ``endpoint`` is positional, not an option). +* Safety bounds on flags that the CLI clamps internally (``--top`` on + ``bcli get`` is hard-capped at 1000 with a default of 50, for + example). + +Both are additive — the existing describe schema is preserved. Older +consumers that ignore these fields keep working. +""" + +from __future__ import annotations + +import json + +from typer.testing import CliRunner + +from bcli_cli.app import app + + +runner = CliRunner() + + +def _describe_json() -> dict: + result = runner.invoke(app, ["describe", "--format", "json"]) + assert result.exit_code == 0, result.stdout + return json.loads(result.stdout) + + +def _find(payload: dict, path: list[str]) -> dict: + matches = [c for c in payload["commands"] if c["path"] == path] + assert matches, f"command {path} not in describe output" + return matches[0] + + +class TestPositionals: + def test_post_has_endpoint_positional(self): + """``bcli post --data ...`` — ``endpoint`` is a + required positional. Describe must surface that so the MCP tool + generator can include it in the tool's input schema.""" + payload = _describe_json() + post = _find(payload, ["post"]) + assert "positionals" in post, post.keys() + names = [p["name"] for p in post["positionals"]] + assert "endpoint" in names + endpoint = next(p for p in post["positionals"] if p["name"] == "endpoint") + assert endpoint["required"] is True + assert endpoint["type"] in {"str", "string"} + + def test_patch_has_endpoint_and_record_id(self): + payload = _describe_json() + patch = _find(payload, ["patch"]) + names = [p["name"] for p in patch["positionals"]] + assert names[:2] == ["endpoint", "record_id"] + assert all(p["required"] for p in patch["positionals"][:2]) + + def test_get_has_optional_record_id(self): + """``bcli get `` vs ``bcli get ``. The id is + positional but optional. Describe records that. + + (The signature names the entity-set positional ``endpoint`` for + historical reasons — same parameter name as post/patch/delete.) + """ + payload = _describe_json() + get_cmd = _find(payload, ["get"]) + positionals = {p["name"]: p for p in get_cmd["positionals"]} + assert "endpoint" in positionals + assert positionals["endpoint"]["required"] is True + if "record_id" in positionals: + assert positionals["record_id"]["required"] is False + + def test_commands_without_positionals_get_empty_list(self): + """``bcli company list`` takes no positional args. The field is + still present (empty) for consistency.""" + payload = _describe_json() + company_list = _find(payload, ["company", "list"]) + assert company_list.get("positionals", []) == [] + + +class TestRequiredOptions: + def test_post_data_option_marked_required(self): + """``bcli post --data ...`` declares ``typer.Option(..., …)`` — + required. Describe surfaces that as ``"required": true`` on the + option entry so the MCP generator can propagate to JSON Schema.""" + payload = _describe_json() + post = _find(payload, ["post"]) + data_opt = next(o for o in post["options"] if o["name"] == "--data") + assert data_opt.get("required") is True + + def test_optional_options_dont_carry_required_flag(self): + """An option with a real default (``typer.Option(False, ...)``) + is NOT required. The flag is omitted, not set to false.""" + payload = _describe_json() + post = _find(payload, ["post"]) + yes_opt = next(o for o in post["options"] if o["name"] == "--yes") + assert "required" not in yes_opt or yes_opt["required"] is False + + +class TestLimits: + def test_get_top_carries_default_and_max(self): + """``bcli get --top`` is a safety-sensitive int. We pin the + default (50 in the CLI) + the hard cap (1000) so an MCP tool + generated from describe can clamp before issuing the call.""" + payload = _describe_json() + get_cmd = _find(payload, ["get"]) + top_opt = next( + (o for o in get_cmd["options"] if o["name"] == "--top"), None, + ) + assert top_opt is not None + # ``limits`` is the new optional sub-object. + limits = top_opt.get("limits", {}) + assert limits.get("default") == 50 + assert limits.get("maximum") == 1000 + # Minimum is whatever the CLI tolerates — 1 in our case. + assert limits.get("minimum", 1) == 1 diff --git a/tests/test_mcp/test_runner.py b/tests/test_mcp/test_runner.py index b226e86..d435900 100644 --- a/tests/test_mcp/test_runner.py +++ b/tests/test_mcp/test_runner.py @@ -13,7 +13,7 @@ import pytest -from bcli_mcp._runner import _strip_rich, run_bcli_json, run_bcli_side_effect +from bcli_mcp._runner import _strip_rich, run_bcli_json, run_bcli_with_envelope from mcp.server.fastmcp.exceptions import ToolError @@ -139,44 +139,77 @@ def test_timeout_raises_toolerror(self): run_bcli_json("get", "customers", timeout=1) -# ── run_bcli_side_effect ───────────────────────────────────────────────── +# ── run_bcli_with_envelope ─────────────────────────────────────────────── -class TestSideEffectRunner: - def test_does_not_append_format_json(self): - """Side-effect commands have non-JSON human stdout — adding - --format json to commands like ``endpoint fields`` would either - be a no-op or change behaviour. Make sure we DON'T append it.""" - with patch("subprocess.run") as run: - run.return_value = _mock_completed(0, stdout="Fields for…") - run_bcli_side_effect("endpoint", "fields", "customers") - argv = run.call_args.args[0] - assert "--format" not in argv +class TestRunBcliWithEnvelope: + """The mutating-tool helper: invoke ``bcli --result-out + --format json``, then read the envelope back. Returns + ``(exit_code, stdout, stderr)`` so the caller decides whether to + treat a non-zero exit as an error (the envelope's ``status`` field + is the true source of truth).""" - def test_returns_none_on_success(self): + def test_appends_result_out_and_format_json(self, tmp_path): + env_path = tmp_path / "out.json" with patch("subprocess.run") as run: - run.return_value = _mock_completed(0, stdout="any text") - assert run_bcli_side_effect("endpoint", "fields", "customers") is None - - def test_ignores_stdout_content(self): - """Non-JSON, garbage, empty — all fine as long as exit is 0.""" + run.return_value = _mock_completed(0, stdout="", stderr="") + rc, _stdout, _stderr = run_bcli_with_envelope( + ["post", "vendors", "--data", "{}"], + env=None, + capture_envelope_path=str(env_path), + ) + argv = run.call_args.args[0] + assert argv[0] == "bcli" + assert "--result-out" in argv + assert argv[argv.index("--result-out") + 1] == str(env_path) + assert "--format" in argv + assert argv[argv.index("--format") + 1] == "json" + assert rc == 0 + + def test_returns_exit_code_unchanged(self, tmp_path): + """Mutating tools care about the envelope's status, not the exit + code per se — but we still pass the exit code through so the + server can sanity-check against the envelope (e.g., if the + envelope is missing, fall back to the exit code).""" with patch("subprocess.run") as run: - run.return_value = _mock_completed(0, stdout="") - run_bcli_side_effect("endpoint", "fields", "customers") # no raise + run.return_value = _mock_completed(6, stdout="", stderr="[red]oops[/red]") + rc, _stdout, stderr = run_bcli_with_envelope( + ["post", "vendors", "--data", "{}"], + env=None, + capture_envelope_path=str(tmp_path / "x.json"), + ) + assert rc == 6 + assert stderr == "[red]oops[/red]" # raw — caller strips as needed - def test_non_zero_exit_raises_toolerror(self): + def test_passes_env_dict_through(self, tmp_path): with patch("subprocess.run") as run: - run.return_value = _mock_completed( - 1, stdout="", stderr="[red]Endpoint not found[/red]", + run.return_value = _mock_completed(0) + run_bcli_with_envelope( + ["post", "v"], + env={"BCLI_PROFILE": "prod", "PATH": "/usr/bin"}, + capture_envelope_path=str(tmp_path / "x.json"), ) - with pytest.raises(ToolError, match=r"Endpoint not found"): - run_bcli_side_effect("endpoint", "fields", "nope") + passed = run.call_args.kwargs["env"] + assert passed["BCLI_PROFILE"] == "prod" - def test_profile_passes_through(self): - with patch("subprocess.run") as run: - run.return_value = _mock_completed(0, stdout="") - run_bcli_side_effect("endpoint", "fields", "customers", profile="prod") - argv = run.call_args.args[0] - env = run.call_args.kwargs["env"] - assert "--profile" in argv and argv[argv.index("--profile") + 1] == "prod" - assert env["BCLI_PROFILE"] == "prod" + def test_missing_bcli_binary_raises_toolerror(self, tmp_path): + with patch("subprocess.run", side_effect=FileNotFoundError("bcli")): + with pytest.raises(ToolError, match=r"bcli executable not found"): + run_bcli_with_envelope( + ["post", "v"], + env=None, + capture_envelope_path=str(tmp_path / "x.json"), + ) + + def test_timeout_raises_toolerror(self, tmp_path): + with patch( + "subprocess.run", + side_effect=subprocess.TimeoutExpired(["bcli"], 1), + ): + with pytest.raises(ToolError, match=r"timed out"): + run_bcli_with_envelope( + ["post", "v"], + env=None, + capture_envelope_path=str(tmp_path / "x.json"), + timeout=1, + ) diff --git a/tests/test_mcp/test_server_tools.py b/tests/test_mcp/test_server_tools.py index 6b4ee3a..ded92bb 100644 --- a/tests/test_mcp/test_server_tools.py +++ b/tests/test_mcp/test_server_tools.py @@ -1,12 +1,23 @@ -"""Tests for the bcli-mcp tool callables. +"""Tests for the bcli-mcp server after the Phase 5 rewrite. -We invoke each ``@mcp.tool()`` function directly with the runner mocked. -Tests assert: argv shape passed to bcli, top-cap enforcement on ``query``, -single-record wrap behaviour, and profile passthrough. +The server now: + +1. Subprocesses ``bcli describe --format json`` once on startup. +2. Builds the tool list from the describe output via + ``_tool_generator.generate_tools``. +3. Registers each generated tool with FastMCP. Read tools shell out via + ``run_bcli_json`` and return parsed stdout. Mutating tools pass + ``--result-out ``, subprocess the CLI, and return the + envelope content as the tool result. + +These tests mock the describe subprocess + the bcli subprocesses so no +real bcli is invoked. """ from __future__ import annotations +import json +from pathlib import Path from unittest.mock import patch import pytest @@ -14,202 +25,304 @@ from bcli_mcp import _server -# ── query ───────────────────────────────────────────────────────────────── - - -class TestQueryTool: +# ── Fixtures ────────────────────────────────────────────────────────── + + +_DESCRIBE_FIXTURE = { + "version": "0.1", + "tool": "bcli", + "tool_version": "0.4.0", + "profile": "dev", + "commands": [ + { + "path": ["get"], + "summary": "GET records from a Business Central entity.", + "options": [ + {"name": "--filter", "type": "str", "validates": "odata-filter"}, + {"name": "--top", "type": "int", + "limits": {"default": 50, "minimum": 1, "maximum": 1000}}, + {"name": "--format", "type": "str"}, + ], + "positionals": [ + {"name": "endpoint", "type": "str", "required": True}, + {"name": "record_id", "type": "str", "required": False}, + ], + "effects": ["read"], + "supported_formats": ["table", "json"], + }, + { + "path": ["post"], + "summary": "POST (create) a new record.", + "options": [ + {"name": "--data", "type": "str"}, + {"name": "--yes", "type": "bool"}, + ], + "positionals": [ + {"name": "endpoint", "type": "str", "required": True}, + ], + "effects": ["mutating"], + "supported_formats": ["json"], + "emits_result_envelope": True, + }, + { + "path": ["company", "list"], + "summary": "List companies on the active environment.", + "options": [], + "positionals": [], + "effects": ["read"], + "supported_formats": ["json"], + }, + ], + "registry": {"tier_1_custom_count": 0, "tier_2_standard_enabled": True, "endpoints": []}, + "profile_constraints": { + "disable_writes": False, "disable_standard_api": False, "allowed_categories": None, + }, +} + + +@pytest.fixture +def fresh_server(): + """Force the server module to rebuild its tool list from a fixture + describe payload. Yields the rebuilt FastMCP instance.""" + instance = _server._build_server(describe_payload=_DESCRIBE_FIXTURE) + yield instance + + +# ── Tool list mirrors describe ──────────────────────────────────────── + + +class TestToolListMirrorsDescribe: + def test_one_tool_per_command_in_describe(self, fresh_server): + """Every command entry → one MCP tool. No extras, no missing.""" + registered = set(fresh_server._tool_manager._tools.keys()) + assert registered == {"bcli_get", "bcli_post", "bcli_company_list"} + + def test_other_effects_excluded(self): + """``effects: ["other"]`` (e.g. ``auth login``, ``config init``) + is filtered out — interactive commands aren't safely scriptable.""" + payload = dict(_DESCRIBE_FIXTURE) + payload["commands"] = list(_DESCRIBE_FIXTURE["commands"]) + [ + {"path": ["auth", "login"], "summary": "", "options": [], + "positionals": [], "effects": ["other"], "supported_formats": ["json"]}, + ] + server = _server._build_server(describe_payload=payload) + assert "bcli_auth_login" not in server._tool_manager._tools + + +# ── Tool invocation: read path ──────────────────────────────────────── + + +class TestReadToolInvocation: @pytest.mark.asyncio - async def test_default_top_is_50(self): - with patch("bcli_mcp._server.run_bcli_json", return_value=[]) as run: - await _server.query(entity="customers") - args = run.call_args.args - assert "--top" in args - assert args[args.index("--top") + 1] == "50" - - @pytest.mark.asyncio - async def test_explicit_top_passes_through(self): - with patch("bcli_mcp._server.run_bcli_json", return_value=[]) as run: - await _server.query(entity="customers", top=200) - args = run.call_args.args - assert args[args.index("--top") + 1] == "200" - - @pytest.mark.asyncio - async def test_top_clamps_at_max(self): - """An agent asking for 50000 records gets clamped — keeps responses - bounded so the model can't accidentally swamp its own context.""" - with patch("bcli_mcp._server.run_bcli_json", return_value=[]) as run: - await _server.query(entity="customers", top=50_000) - args = run.call_args.args - assert args[args.index("--top") + 1] == "1000" - - @pytest.mark.asyncio - async def test_top_floor_is_1(self): - with patch("bcli_mcp._server.run_bcli_json", return_value=[]) as run: - await _server.query(entity="customers", top=0) - args = run.call_args.args - assert args[args.index("--top") + 1] == "1" - - @pytest.mark.asyncio - async def test_filter_select_orderby_passed_through(self): - with patch("bcli_mcp._server.run_bcli_json", return_value=[]) as run: - await _server.query( - entity="customers", - filter="displayName eq 'X'", - select="number,displayName", - orderby="number desc", - ) + async def test_read_tool_subprocesses_with_format_json(self, fresh_server): + """A read tool call → ``bcli get customers --filter ... --format json``.""" + tool = fresh_server._tool_manager._tools["bcli_get"] + with patch( + "bcli_mcp._server.run_bcli_json", + return_value=[{"id": "c-1"}], + ) as run: + result = await tool.fn(endpoint="customers", filter="x eq 1") args = run.call_args.args + # Positional first, then option flags. + assert args[0:2] == ("get", "customers") assert "--filter" in args - assert args[args.index("--filter") + 1] == "displayName eq 'X'" - assert args[args.index("--select") + 1] == "number,displayName" - assert args[args.index("--orderby") + 1] == "number desc" - - @pytest.mark.asyncio - async def test_record_id_appended_after_entity(self): - with patch("bcli_mcp._server.run_bcli_json", return_value={"id": "c-1"}) as run: - result = await _server.query(entity="customers", record_id="c-1") - args = run.call_args.args - assert args[:3] == ("get", "customers", "c-1") - # Single-record dict gets wrapped to keep return type stable + assert args[args.index("--filter") + 1] == "x eq 1" assert result == [{"id": "c-1"}] @pytest.mark.asyncio - async def test_publisher_group_version_passed_through(self): - with patch("bcli_mcp._server.run_bcli_json", return_value=[]) as run: - await _server.query( - entity="customEntity", - publisher="acme", group="finance", version="v1.5", - ) - args = run.call_args.args - assert args[args.index("--publisher") + 1] == "acme" - assert args[args.index("--group") + 1] == "finance" - assert args[args.index("--version") + 1] == "v1.5" + async def test_read_tool_with_only_required_positional(self, fresh_server): + tool = fresh_server._tool_manager._tools["bcli_company_list"] + with patch( + "bcli_mcp._server.run_bcli_json", + return_value=[{"id": "BTUSALLC"}], + ) as run: + result = await tool.fn() + assert run.call_args.args == ("company", "list") + assert result == [{"id": "BTUSALLC"}] @pytest.mark.asyncio - async def test_profile_passes_to_runner(self): - with patch("bcli_mcp._server.run_bcli_json", return_value=[]) as run: - await _server.query(entity="customers", profile="sandbox") - assert run.call_args.kwargs["profile"] == "sandbox" + async def test_profile_kwarg_passes_through(self, fresh_server): + tool = fresh_server._tool_manager._tools["bcli_get"] + with patch( + "bcli_mcp._server.run_bcli_json", return_value=[], + ) as run: + await tool.fn(endpoint="customers", profile="prod") + assert run.call_args.kwargs["profile"] == "prod" -# ── list_endpoints ──────────────────────────────────────────────────────── +# ── Tool invocation: mutating path ──────────────────────────────────── -class TestListEndpointsTool: +class TestMutatingToolInvocation: @pytest.mark.asyncio - async def test_basic_call(self): - with patch("bcli_mcp._server.run_bcli_json", return_value=[]) as run: - await _server.list_endpoints() - assert run.call_args.args == ("endpoint", "list") + async def test_mutating_tool_passes_result_out_and_reads_envelope( + self, fresh_server, tmp_path, + ): + """The server must pass ``--result-out `` and return the + envelope content. Stdout is dropped (envelope has everything).""" + envelope = { + "version": "0.1", "invocation_id": "inv-1", "tool_version": "0.4.0", + "profile": "dev", "environment": "Sandbox", "company": "c-1", + "method": "POST", "endpoint": "vendors", + "resolved_url": "https://x/vendors", "record_id": "vnd-9", + "dry_run": False, "status": "succeeded", "exit_code": 0, + "bc_correlation_id": None, "telemetry_event_id": None, + "audit_log_offset": None, "started_at": "2026-05-17T00:00:00Z", + "duration_ms": 42, + } - @pytest.mark.asyncio - async def test_filters_passed_through(self): - with patch("bcli_mcp._server.run_bcli_json", return_value=[]) as run: - await _server.list_endpoints( - category="finance", custom_only=True, - ) - args = run.call_args.args - assert "--custom" in args - assert args[args.index("--category") + 1] == "finance" + def _fake_run(argv, env, capture_envelope_path): + # Write the envelope where the runner expects it. + Path(capture_envelope_path).write_text(json.dumps(envelope)) + return 0, "", "" + + tool = fresh_server._tool_manager._tools["bcli_post"] + with patch("bcli_mcp._server.run_bcli_with_envelope", side_effect=_fake_run): + result = await tool.fn(endpoint="vendors", data='{"x":1}', yes=True) + assert result == envelope @pytest.mark.asyncio - async def test_standard_only_flag(self): - with patch("bcli_mcp._server.run_bcli_json", return_value=[]) as run: - await _server.list_endpoints(standard_only=True) - assert "--standard" in run.call_args.args + async def test_failed_envelope_raises_toolerror(self, fresh_server, tmp_path): + """When the envelope's ``status="failed"``, the tool surfaces an + MCP error with the envelope as the body so the agent can read + ``exit_code``, ``bc_correlation_id``, etc.""" + from mcp.server.fastmcp.exceptions import ToolError + failed = { + "version": "0.1", "invocation_id": "inv-2", "tool_version": "0.4.0", + "profile": "dev", "environment": "Sandbox", "company": "c-1", + "method": "POST", "endpoint": "vendors", "resolved_url": None, + "record_id": None, "dry_run": False, "status": "failed", + "exit_code": 6, "bc_correlation_id": "corr-xyz", + "telemetry_event_id": None, "audit_log_offset": None, + "started_at": "2026-05-17T00:00:00Z", "duration_ms": 11, + } -# ── describe_endpoint ───────────────────────────────────────────────────── + def _fake_run(argv, env, capture_envelope_path): + Path(capture_envelope_path).write_text(json.dumps(failed)) + return 6, "", "" + tool = fresh_server._tool_manager._tools["bcli_post"] + with patch("bcli_mcp._server.run_bcli_with_envelope", side_effect=_fake_run): + with pytest.raises(ToolError) as excinfo: + await tool.fn(endpoint="vendors", data='{"x":1}', yes=True) + # The exception message includes the correlation id so the agent + # can quote it back when raising a support ticket. + assert "corr-xyz" in str(excinfo.value) -class TestDescribeEndpointTool: @pytest.mark.asyncio - async def test_calls_endpoint_info(self): - with patch( - "bcli_mcp._server.run_bcli_json", - return_value={"name": "customers"}, - ) as run: - result = await _server.describe_endpoint(name="customers") - assert run.call_args.args == ("endpoint", "info", "customers") - assert result == {"name": "customers"} + async def test_mutating_tool_passes_clean_argv_to_runner( + self, fresh_server, tmp_path, + ): + """The server hands the runner a CLEAN argv (positionals + flags + only). ``--result-out`` and ``--format json`` are appended inside + :func:`run_bcli_with_envelope` so the boundary is single-sourced. + + The runner-side contract is exercised in ``test_runner.py``; + here we just pin the server side: it shouldn't be re-adding + either flag itself.""" + envelope = { + "version": "0.1", "invocation_id": "x", "tool_version": "0.4.0", + "profile": "dev", "environment": "Sandbox", "company": "c-1", + "method": "POST", "endpoint": "vendors", "resolved_url": None, + "record_id": None, "dry_run": False, "status": "succeeded", + "exit_code": 0, "bc_correlation_id": None, + "telemetry_event_id": None, "audit_log_offset": None, + "started_at": "x", "duration_ms": 0, + } + captured_argv = {} - @pytest.mark.asyncio - async def test_discover_fields_default_false_skips_side_effect(self): - with patch("bcli_mcp._server.run_bcli_json", return_value={}) as info, \ - patch("bcli_mcp._server.run_bcli_side_effect") as side: - await _server.describe_endpoint(name="customers") - side.assert_not_called() - info.assert_called_once() + def _fake_run(argv, env, capture_envelope_path): + captured_argv["v"] = argv + Path(capture_envelope_path).write_text(json.dumps(envelope)) + return 0, "", "" - @pytest.mark.asyncio - async def test_discover_fields_true_runs_fields_then_info(self): - with patch("bcli_mcp._server.run_bcli_json", return_value={}) as info, \ - patch("bcli_mcp._server.run_bcli_side_effect") as side: - await _server.describe_endpoint(name="customers", discover_fields=True) - side.assert_called_once_with( - "endpoint", "fields", "customers", profile=None, - ) - info.assert_called_once_with( - "endpoint", "info", "customers", profile=None, - ) + tool = fresh_server._tool_manager._tools["bcli_post"] + with patch("bcli_mcp._server.run_bcli_with_envelope", side_effect=_fake_run): + await tool.fn(endpoint="vendors", data="{}", yes=True) + argv = captured_argv["v"] + # Positional first, options after. + assert argv[:2] == ["post", "vendors"] + assert "--data" in argv + assert argv[argv.index("--data") + 1] == "{}" + # The server passes a clean argv — runner adds --result-out / --format. + assert "--result-out" not in argv + assert "--format" not in argv - @pytest.mark.asyncio - async def test_discover_fields_swallows_side_effect_failure(self): - """If discovery fails (entity needs filter, no records, etc.) we - still return the info payload — fields_discovered will be False - and the agent can fall back to a probe query.""" - from mcp.server.fastmcp.exceptions import ToolError - with patch( - "bcli_mcp._server.run_bcli_json", - return_value={"name": "customerSales", "fields_discovered": False}, - ) as info, patch( - "bcli_mcp._server.run_bcli_side_effect", - side_effect=ToolError("BC returned 400"), - ): - result = await _server.describe_endpoint( - name="customerSales", discover_fields=True, - ) - assert result == {"name": "customerSales", "fields_discovered": False} - info.assert_called_once() +# ── Profile passthrough on mutating tools ───────────────────────────── + +class TestMutatingProfilePassthrough: @pytest.mark.asyncio - async def test_discover_fields_passes_profile_through(self): - with patch("bcli_mcp._server.run_bcli_json", return_value={}), \ - patch("bcli_mcp._server.run_bcli_side_effect") as side: - await _server.describe_endpoint( - name="customers", discover_fields=True, profile="sandbox", - ) - assert side.call_args.kwargs["profile"] == "sandbox" + async def test_profile_propagates_to_subprocess(self, fresh_server, tmp_path): + envelope = { + "version": "0.1", "invocation_id": "x", "tool_version": "0.4.0", + "profile": "prod", "environment": "Production", "company": "c-1", + "method": "POST", "endpoint": "v", "resolved_url": None, + "record_id": None, "dry_run": False, "status": "succeeded", + "exit_code": 0, "bc_correlation_id": None, + "telemetry_event_id": None, "audit_log_offset": None, + "started_at": "x", "duration_ms": 0, + } + captured = {} + def _fake_run(argv, env, capture_envelope_path): + captured["argv"] = argv + captured["env"] = env + Path(capture_envelope_path).write_text(json.dumps(envelope)) + return 0, "", "" -# ── list_companies ──────────────────────────────────────────────────────── + tool = fresh_server._tool_manager._tools["bcli_post"] + with patch("bcli_mcp._server.run_bcli_with_envelope", side_effect=_fake_run): + await tool.fn(endpoint="v", data="{}", yes=True, profile="prod") + assert "--profile" in captured["argv"] + assert captured["argv"][captured["argv"].index("--profile") + 1] == "prod" + assert captured["env"]["BCLI_PROFILE"] == "prod" -class TestListCompaniesTool: - @pytest.mark.asyncio - async def test_calls_company_list(self): - with patch("bcli_mcp._server.run_bcli_json", return_value=[]) as run: - await _server.list_companies() - assert run.call_args.args == ("company", "list") +# ── Describe failure fallback ───────────────────────────────────────── - @pytest.mark.asyncio - async def test_profile_propagates(self): - with patch("bcli_mcp._server.run_bcli_json", return_value=[]) as run: - await _server.list_companies(profile="sandbox") - assert run.call_args.kwargs["profile"] == "sandbox" +class TestDescribeFailureFallback: + def test_describe_failure_yields_empty_tool_list(self): + """If ``bcli describe`` exits non-zero or doesn't exist, the + server doesn't crash on startup — it builds with zero tools and + logs the error. Caller can re-invoke after fixing the install.""" + with patch( + "bcli_mcp._server._load_describe_payload", + side_effect=RuntimeError("bcli not found"), + ): + server = _server._build_server() # no payload kwarg + # No tools registered, but the server itself is built. + assert server._tool_manager._tools == {} + + def test_describe_returns_no_commands_yields_empty_tool_list(self): + """An empty ``commands: []`` (broken install? unconfigured profile?) + is treated the same — start clean, register nothing.""" + empty = { + "version": "0.1", "tool": "bcli", "tool_version": "0.4.0", + "profile": None, "commands": [], + "registry": {"tier_1_custom_count": 0, "tier_2_standard_enabled": True, + "endpoints": []}, + "profile_constraints": {"disable_writes": None, + "disable_standard_api": None, + "allowed_categories": None}, + } + server = _server._build_server(describe_payload=empty) + assert server._tool_manager._tools == {} -# ── Tool registration sanity ────────────────────────────────────────────── +# ── Tools generated only once on startup ────────────────────────────── -class TestServerRegistration: - def test_four_tools_exposed(self): - """The tool surface is deliberately small — the design contract.""" - registered = set(_server.mcp._tool_manager._tools.keys()) - assert registered == { - "query", - "list_endpoints", - "describe_endpoint", - "list_companies", - } + +class TestStartupCache: + def test_describe_invoked_once_per_build(self): + """``_build_server`` calls describe at most once; later tool + calls re-use the cached payload + generated tools.""" + with patch( + "bcli_mcp._server._load_describe_payload", + return_value=_DESCRIBE_FIXTURE, + ) as load: + _server._build_server() + assert load.call_count == 1 diff --git a/tests/test_mcp/test_tool_generator.py b/tests/test_mcp/test_tool_generator.py new file mode 100644 index 0000000..b60fe3e --- /dev/null +++ b/tests/test_mcp/test_tool_generator.py @@ -0,0 +1,395 @@ +"""Tests for the dynamic tool generator. + +The generator parses ``bcli describe --format json`` output and yields +one ``GeneratedTool`` per command entry. Mutating tools carry the +envelope contract: when invoked, the server must pass ``--result-out`` +and return the envelope content as the tool result. + +These tests cover the generator in isolation — no MCP server, no real +subprocess. Server integration is exercised in ``test_server_tools.py``. +""" + +from __future__ import annotations + +import pytest + +from bcli_mcp._tool_generator import ( + _option_to_input_property, + _path_to_tool_name, + generate_tools, +) + + +# ── Fixtures ────────────────────────────────────────────────────────── + + +def _describe_payload(*commands: dict) -> dict: + """Minimal describe payload with the given commands.""" + return { + "version": "0.1", + "tool": "bcli", + "tool_version": "0.4.0", + "profile": "dev", + "commands": list(commands), + "registry": { + "tier_1_custom_count": 0, + "tier_2_standard_enabled": True, + "endpoints": [], + }, + "profile_constraints": { + "disable_writes": False, + "disable_standard_api": False, + "allowed_categories": None, + }, + } + + +# ── _path_to_tool_name ──────────────────────────────────────────────── + + +class TestPathToToolName: + def test_single_word_path(self): + assert _path_to_tool_name(["get"]) == "bcli_get" + + def test_two_word_path(self): + assert _path_to_tool_name(["batch", "run"]) == "bcli_batch_run" + + def test_three_word_path(self): + assert _path_to_tool_name(["attach", "upload"]) == "bcli_attach_upload" + + def test_hyphens_become_underscores(self): + """MCP tool names must be a valid identifier so hyphens flatten.""" + assert _path_to_tool_name(["ai-context"]) == "bcli_ai_context" + assert _path_to_tool_name(["batch", "list-templates"]) == "bcli_batch_list_templates" + + +# ── _option_to_input_property ───────────────────────────────────────── + + +class TestOptionToInputProperty: + def test_string_option(self): + prop = _option_to_input_property({"name": "--filter", "type": "str"}) + assert prop == {"type": "string"} + + def test_int_option(self): + prop = _option_to_input_property({"name": "--top", "type": "int"}) + assert prop == {"type": "integer"} + + def test_bool_option(self): + prop = _option_to_input_property({"name": "--yes", "type": "bool"}) + assert prop == {"type": "boolean"} + + def test_path_option(self): + # Path falls through to string for JSON Schema purposes. + prop = _option_to_input_property({"name": "--output", "type": "path"}) + assert prop == {"type": "string"} + + def test_unknown_type_falls_back_to_string(self): + prop = _option_to_input_property({"name": "--mystery", "type": "weirdo"}) + assert prop == {"type": "string"} + + def test_limits_emitted_when_present(self): + """Safety limits from describe carry into the JSON schema so an + agent runtime can clamp before invoking the tool.""" + prop = _option_to_input_property({ + "name": "--top", + "type": "int", + "limits": {"default": 50, "minimum": 1, "maximum": 1000}, + }) + assert prop["type"] == "integer" + assert prop["default"] == 50 + assert prop["minimum"] == 1 + assert prop["maximum"] == 1000 + + +# ── generate_tools — read commands ──────────────────────────────────── + + +class TestGenerateReadTools: + def test_read_command_becomes_one_tool(self): + payload = _describe_payload({ + "path": ["get"], + "summary": "GET records from a Business Central entity.", + "options": [ + {"name": "--filter", "type": "str", "validates": "odata-filter"}, + {"name": "--top", "type": "int", "limits": {"default": 50, "maximum": 1000}}, + ], + "positionals": [ + {"name": "endpoint", "type": "str", "required": True}, + {"name": "record_id", "type": "str", "required": False}, + ], + "effects": ["read"], + "supported_formats": ["json", "table"], + }) + tools = generate_tools(payload) + assert len(tools) == 1 + t = tools[0] + assert t.name == "bcli_get" + assert t.effects == ["read"] + assert t.emits_envelope is False + # Schema captures both positionals and options. + assert "endpoint" in t.input_schema["properties"] + assert "record_id" in t.input_schema["properties"] + assert "filter" in t.input_schema["properties"] + assert "top" in t.input_schema["properties"] + # Required list matches positionals[required=True]. + assert t.input_schema["required"] == ["endpoint"] + # Safety limit propagated. + assert t.input_schema["properties"]["top"]["maximum"] == 1000 + + def test_summary_becomes_description(self): + payload = _describe_payload({ + "path": ["company", "list"], + "summary": "List companies on the active environment.", + "options": [], + "effects": ["read"], + "supported_formats": ["json"], + }) + tools = generate_tools(payload) + assert tools[0].description == "List companies on the active environment." + + +# ── generate_tools — mutating commands ──────────────────────────────── + + +class TestGenerateMutatingTools: + def test_required_option_marked_required_in_schema(self): + """``--data`` on ``bcli post`` is ``typer.Option(..., …)`` — i.e. + required. Describe surfaces ``required: true``; the generator + propagates that into JSON Schema's ``required`` array so an + agent doesn't construct a missing-argument call.""" + payload = _describe_payload({ + "path": ["post"], + "summary": "", + "options": [ + {"name": "--data", "type": "str", "required": True}, + {"name": "--yes", "type": "bool"}, + ], + "positionals": [ + {"name": "endpoint", "type": "str", "required": True}, + ], + "effects": ["mutating"], + "supported_formats": ["json"], + "emits_result_envelope": True, + }) + t = generate_tools(payload)[0] + # Both the positional ``endpoint`` and the required option + # ``data`` end up in the schema's required list. + assert set(t.input_schema["required"]) == {"endpoint", "data"} + + def test_mutating_command_emits_envelope_flag(self): + payload = _describe_payload({ + "path": ["post"], + "summary": "POST (create) a new record.", + "options": [ + {"name": "--data", "type": "str"}, + {"name": "--yes", "type": "bool"}, + ], + "positionals": [ + {"name": "endpoint", "type": "str", "required": True}, + ], + "effects": ["mutating"], + "supported_formats": ["json"], + "emits_result_envelope": True, + "requires_confirmation": "production", + }) + tools = generate_tools(payload) + assert len(tools) == 1 + t = tools[0] + assert t.name == "bcli_post" + assert t.effects == ["mutating"] + assert t.emits_envelope is True + # The agent sees the same input shape as the CLI accepts. + props = t.input_schema["properties"] + assert "endpoint" in props + assert "data" in props + assert "yes" in props + assert t.input_schema["required"] == ["endpoint"] + + def test_envelope_emitted_only_when_declared(self): + """A command flagged ``effects: ["mutating"]`` without + ``emits_result_envelope: true`` doesn't get the envelope wiring — + we don't infer it from effects alone.""" + payload = _describe_payload({ + "path": ["weird"], + "summary": "", + "options": [], + "effects": ["mutating"], + "supported_formats": ["json"], + # emits_result_envelope NOT set + }) + t = generate_tools(payload)[0] + assert t.emits_envelope is False + + +# ── generate_tools — all together ───────────────────────────────────── + + +class TestGenerateAllCommands: + def test_tools_count_matches_command_count(self): + payload = _describe_payload( + {"path": ["get"], "summary": "", "options": [], "effects": ["read"], + "supported_formats": ["json"]}, + {"path": ["post"], "summary": "", "options": [], "effects": ["mutating"], + "supported_formats": ["json"], "emits_result_envelope": True}, + {"path": ["company", "list"], "summary": "", "options": [], + "effects": ["read"], "supported_formats": ["json"]}, + ) + tools = generate_tools(payload) + assert {t.name for t in tools} == {"bcli_get", "bcli_post", "bcli_company_list"} + + def test_invocations_skipped_for_other_effects(self): + """Commands with effects=["other"] (e.g. ``config init``, ``auth + login``) are surfaced read-style but never as mutating tools. + Excluding them keeps the surface tight and avoids accidentally + wrapping interactive prompts.""" + payload = _describe_payload( + {"path": ["auth", "login"], "summary": "", "options": [], + "effects": ["other"], "supported_formats": ["json"]}, + ) + # Default policy: skip "other". + tools = generate_tools(payload) + assert tools == [] + + +# ── invocation arg-building ────────────────────────────────────────── + + +class TestBuildInvocationArgs: + def test_read_tool_builds_get_argv(self): + payload = _describe_payload({ + "path": ["get"], + "summary": "", + "options": [ + {"name": "--filter", "type": "str"}, + {"name": "--top", "type": "int"}, + ], + "positionals": [ + {"name": "endpoint", "type": "str", "required": True}, + ], + "effects": ["read"], + "supported_formats": ["json"], + }) + t = generate_tools(payload)[0] + args = t.build_argv({"endpoint": "customers", "filter": "x eq 1", "top": 10}) + # Positionals come first in path order. + assert args[0:2] == ["get", "customers"] + # Option flags follow as --name value pairs. + assert "--filter" in args + assert args[args.index("--filter") + 1] == "x eq 1" + assert args[args.index("--top") + 1] == "10" + + def test_skips_missing_optional_args(self): + payload = _describe_payload({ + "path": ["get"], + "summary": "", + "options": [ + {"name": "--filter", "type": "str"}, + {"name": "--top", "type": "int"}, + ], + "positionals": [ + {"name": "endpoint", "type": "str", "required": True}, + ], + "effects": ["read"], + "supported_formats": ["json"], + }) + t = generate_tools(payload)[0] + args = t.build_argv({"endpoint": "customers"}) + assert "--filter" not in args + assert "--top" not in args + + def test_bool_flag_only_when_true(self): + payload = _describe_payload({ + "path": ["post"], + "summary": "", + "options": [ + {"name": "--data", "type": "str"}, + {"name": "--yes", "type": "bool"}, + ], + "positionals": [ + {"name": "endpoint", "type": "str", "required": True}, + ], + "effects": ["mutating"], + "supported_formats": ["json"], + "emits_result_envelope": True, + }) + t = generate_tools(payload)[0] + args_true = t.build_argv({"endpoint": "vendors", "data": "{}", "yes": True}) + assert "--yes" in args_true + args_false = t.build_argv({"endpoint": "vendors", "data": "{}", "yes": False}) + assert "--yes" not in args_false + + def test_list_positional_splits_whitespace_string(self): + """``bcli describe`` accepts a list-of-strings positional. The MCP + surface exposes it as a single string; build_argv splits at call + time so the subprocess sees the right argv. Without this, an + agent calling ``bcli_describe(command_path="batch run")`` would + produce ``bcli describe "batch run"`` (one quoted token) — the + CLI 4xxs because it expects two tokens.""" + payload = _describe_payload({ + "path": ["describe"], + "summary": "", + "options": [], + "positionals": [ + {"name": "command_path", "type": "list[str]", "required": False}, + ], + "effects": ["read"], + "supported_formats": ["json"], + }) + t = generate_tools(payload)[0] + args = t.build_argv({"command_path": "batch run"}) + assert args == ["describe", "batch", "run"] + + def test_list_positional_accepts_actual_list(self): + payload = _describe_payload({ + "path": ["describe"], + "summary": "", + "options": [], + "positionals": [ + {"name": "command_path", "type": "list[str]", "required": False}, + ], + "effects": ["read"], + "supported_formats": ["json"], + }) + t = generate_tools(payload)[0] + args = t.build_argv({"command_path": ["batch", "run"]}) + assert args == ["describe", "batch", "run"] + + def test_positional_required_validated(self): + """build_argv raises if a required positional is missing — better + to fail in the wrapper than send malformed argv to bcli.""" + payload = _describe_payload({ + "path": ["post"], + "summary": "", + "options": [{"name": "--data", "type": "str"}], + "positionals": [ + {"name": "endpoint", "type": "str", "required": True}, + ], + "effects": ["mutating"], + "supported_formats": ["json"], + "emits_result_envelope": True, + }) + t = generate_tools(payload)[0] + with pytest.raises(ValueError, match=r"endpoint"): + t.build_argv({"data": "{}"}) + + +# ── Edge case: no positionals field on the describe entry ──────────── + + +class TestBackwardsCompatibility: + def test_describe_without_positionals_field_is_tolerated(self): + """An older describe output (pre-Phase 5) doesn't carry + ``positionals``. Treat missing as an empty list so old caches + and older bcli versions still produce a usable tool list.""" + payload = _describe_payload({ + "path": ["get"], + "summary": "GET records.", + "options": [{"name": "--top", "type": "int"}], + # no positionals key + "effects": ["read"], + "supported_formats": ["json"], + }) + tools = generate_tools(payload) + assert len(tools) == 1 + assert tools[0].input_schema["required"] == []