diff --git a/GET_STARTED.md b/GET_STARTED.md index c2dee7b..356a46a 100644 --- a/GET_STARTED.md +++ b/GET_STARTED.md @@ -27,6 +27,96 @@ SpecLeft is passive by default — it won't modify anything until you ask. --- +## MCP Server Setup + +SpecLeft includes an MCP server that connects directly to AI coding agents like +Claude Code, Cursor, and Windsurf. Once connected, your agent can read specs, +track coverage, and generate test scaffolding without leaving the conversation. + +### Prerequisites + +- Python 3.10+ +- SpecLeft installed (`pip install specleft[mcp]`) + +### Option 1: uvx (recommended) + +If you have [uv](https://docs.astral.sh/uv/getting-started/installation/) +installed, this is the fastest path. No separate install step is required. + +**Claude Code** (`.claude/settings.json`): + +```json +{ + "mcpServers": { + "specleft": { + "command": "uvx", + "args": ["specleft", "mcp"] + } + } +} +``` + +**Cursor** (`.cursor/mcp.json`): + +```json +{ + "mcpServers": { + "specleft": { + "command": "uvx", + "args": ["specleft", "mcp"] + } + } +} +``` + +### Option 2: pip install + +If you do not have `uv`, install SpecLeft first and point your MCP client at +the CLI directly: + +```bash +pip install specleft[mcp] +``` + +**Claude Code** (`.claude/settings.json`): + +```json +{ + "mcpServers": { + "specleft": { + "command": "specleft", + "args": ["mcp"] + } + } +} +``` + +**Cursor** (`.cursor/mcp.json`): + +```json +{ + "mcpServers": { + "specleft": { + "command": "specleft", + "args": ["mcp"] + } + } +} +``` + +### Verify the connection + +After configuration, restart your editor. You should see SpecLeft listed as a +connected MCP server with 3 resources and 1 tool. If the connection fails, run: + +```bash +specleft doctor +``` + +This checks Python version, dependencies, and plugin registration. + +--- + ## 1. Write a PRD Create a `prd.md` file in your repository root: diff --git a/README.md b/README.md index 04a9dff..e68e263 100644 --- a/README.md +++ b/README.md @@ -158,6 +158,16 @@ pip install specleft No config files required. No test changes required. +--- +## MCP Server Setup + +SpecLeft includes an MCP server that connects directly to AI coding agents +like Claude Code, Cursor, Codex, and OpenCode. Once connected, your agent can +read specs, track coverage, and generate test scaffolding without leaving the +conversation. + +See [GET_STARTED.md](https://github.com/SpecLeft/specleft/blob/main/GET_STARTED.md) for details. + --- ## SpecLeft Agent Contract diff --git a/features/feature-mcp-server.md b/features/feature-mcp-server.md new file mode 100644 index 0000000..18db394 --- /dev/null +++ b/features/feature-mcp-server.md @@ -0,0 +1,38 @@ +# Feature: MCP Server for Agent Discovery +priority: high + +## Scenarios + +### Scenario: Expose exactly three resources and one tool +- Given the SpecLeft MCP server is running +- When an MCP client lists server resources and tools +- Then the server exposes resources `specleft://contract`, `specleft://guide`, and `specleft://status` +- And the server exposes exactly one tool named `specleft_init` + +### Scenario: Contract and guide resources return machine-readable JSON +- Given the SpecLeft MCP server is running +- When an MCP client reads `specleft://contract` and `specleft://guide` +- Then both resources return valid JSON payloads +- And the contract payload includes safety and determinism guarantees +- And the guide payload includes workflow steps and skill file guidance + +### Scenario: Status resource signals uninitialised project +- Given an empty workspace with no SpecLeft setup +- When an MCP client reads `specleft://status` +- Then the payload includes `initialised: false` +- And feature and scenario counts are zero + +### Scenario: specleft_init bootstraps project safely +- Given an empty workspace with write permissions +- When an MCP client calls the `specleft_init` tool +- Then the tool runs health checks before writing files +- And it creates `.specleft/specs`, `.specleft/policies`, and `.specleft/SKILL.md` +- And repeated calls are idempotent + +### Scenario: Agent discovery flow uses resources and init tool +- Given an agent connects to the MCP server +- When the agent reads contract, guide, and status resources +- And status reports `initialised: false` +- And the agent calls `specleft_init` +- Then the workspace is initialised +- And a subsequent status read reports `initialised: true` diff --git a/pyproject.toml b/pyproject.toml index d0ce75c..4764493 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -64,10 +64,14 @@ Documentation = "https://github.com/SpecLeft/specleft/tree/main/docs" Repository = "https://github.com/SpecLeft/specleft" [project.optional-dependencies] +mcp = [ + "fastmcp<3", +] dev = [ "pytest-cov", "pytest-subtests", "pytest-asyncio", + "tiktoken", "black==26.1.0", "ruff==0.8.3", "mypy", diff --git a/src/specleft/cli/main.py b/src/specleft/cli/main.py index 24f174b..16e26c5 100644 --- a/src/specleft/cli/main.py +++ b/src/specleft/cli/main.py @@ -16,6 +16,7 @@ guide, init, license_group, + mcp, next_command, plan, skill_group, @@ -53,6 +54,7 @@ def cli() -> None: cli.add_command(license_group) cli.add_command(skill_group) cli.add_command(guide) +cli.add_command(mcp) __all__ = ["cli"] diff --git a/src/specleft/commands/__init__.py b/src/specleft/commands/__init__.py index 370363c..c6ded42 100644 --- a/src/specleft/commands/__init__.py +++ b/src/specleft/commands/__init__.py @@ -13,6 +13,7 @@ from specleft.commands.guide import guide from specleft.commands.init import init from specleft.commands.license import license_group +from specleft.commands.mcp import mcp from specleft.commands.next import next_command from specleft.commands.plan import plan from specleft.commands.skill import skill_group @@ -28,6 +29,7 @@ "guide", "init", "license_group", + "mcp", "next_command", "plan", "skill_group", diff --git a/src/specleft/commands/mcp.py b/src/specleft/commands/mcp.py new file mode 100644 index 0000000..8d4e619 --- /dev/null +++ b/src/specleft/commands/mcp.py @@ -0,0 +1,23 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026 SpecLeft Contributors + +"""MCP command.""" + +from __future__ import annotations + +import click + +from specleft.utils.messaging import print_support_footer + + +@click.command("mcp") +def mcp() -> None: + """Run the SpecLeft MCP server over stdio.""" + try: + from specleft.mcp.server import run_mcp_server + + run_mcp_server() + except RuntimeError as exc: + click.secho(str(exc), fg="red", err=True) + print_support_footer() + raise SystemExit(1) from exc diff --git a/src/specleft/commands/status.py b/src/specleft/commands/status.py index 84ceb43..5e120c1 100644 --- a/src/specleft/commands/status.py +++ b/src/specleft/commands/status.py @@ -188,7 +188,7 @@ def build_status_json( "coverage_percent": summary.coverage_percent, } if not verbose: - return summary_payload + return {"initialised": True, **summary_payload} features: list[dict[str, Any]] = [] by_priority: dict[str, dict[str, int]] = {} diff --git a/src/specleft/mcp/__init__.py b/src/specleft/mcp/__init__.py new file mode 100644 index 0000000..f7279c9 --- /dev/null +++ b/src/specleft/mcp/__init__.py @@ -0,0 +1,8 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026 SpecLeft Contributors + +"""SpecLeft MCP server package.""" + +from specleft.mcp.server import build_mcp_server, run_mcp_server + +__all__ = ["build_mcp_server", "run_mcp_server"] diff --git a/src/specleft/mcp/__main__.py b/src/specleft/mcp/__main__.py new file mode 100644 index 0000000..2e93841 --- /dev/null +++ b/src/specleft/mcp/__main__.py @@ -0,0 +1,22 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026 SpecLeft Contributors + +"""Entrypoint for ``python -m specleft.mcp``.""" + +from __future__ import annotations + +import sys + +from specleft.mcp.server import run_mcp_server + + +def main() -> None: + try: + run_mcp_server() + except RuntimeError as exc: + print(str(exc), file=sys.stderr) + raise SystemExit(1) from exc + + +if __name__ == "__main__": + main() diff --git a/src/specleft/mcp/init_tool.py b/src/specleft/mcp/init_tool.py new file mode 100644 index 0000000..bb30e93 --- /dev/null +++ b/src/specleft/mcp/init_tool.py @@ -0,0 +1,182 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026 SpecLeft Contributors + +"""Implementation helpers for the MCP ``specleft_init`` tool.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from specleft.commands.doctor import _build_doctor_checks, _build_doctor_output +from specleft.commands.init import _apply_init_plan, _init_plan +from specleft.utils.skill_integrity import ( + SKILL_FILE_PATH, + SKILL_HASH_PATH, + sync_skill_files, +) + + +class SecurityError(RuntimeError): + """Raised when MCP init detects an unsafe write target.""" + + +def _is_relative_to(path: Path, root: Path) -> bool: + try: + path.relative_to(root) + return True + except ValueError: + return False + + +def ensure_safe_write_target(path: Path, workspace: Path | None = None) -> Path: + """Validate that write targets stay within the current workspace.""" + root = (workspace or Path.cwd()).resolve() + candidate = path if path.is_absolute() else (root / path) + + if ".." in path.parts: + raise SecurityError(f"Path traversal is not allowed: {path}") + + if not _is_relative_to(candidate, root): + raise SecurityError(f"Path escapes workspace: {path}") + + current = candidate + while True: + if current.exists() and current.is_symlink(): + raise SecurityError(f"Refusing to write through symlink: {current}") + if current == root: + break + if not _is_relative_to(current, root): + raise SecurityError(f"Path escapes workspace: {path}") + current = current.parent + + resolved_parent = candidate.parent.resolve(strict=False) + if not _is_relative_to(resolved_parent, root): + raise SecurityError(f"Path escapes workspace: {path}") + + if candidate.exists(): + resolved_candidate = candidate.resolve(strict=False) + if not _is_relative_to(resolved_candidate, root): + raise SecurityError(f"Path escapes workspace: {path}") + + return candidate + + +def _health_payload(checks: dict[str, Any]) -> dict[str, Any]: + checks_map = checks.get("checks", {}) + python_check = checks_map.get("python_version", {}) + dependencies_check = checks_map.get("dependencies", {}) + plugin_check = checks_map.get("pytest_plugin", {}) + + return { + "python_version": python_check.get("version"), + "dependencies_ok": dependencies_check.get("status") == "pass", + "plugin_registered": bool(plugin_check.get("registered", False)), + } + + +def _normalize_options(*, example: bool, blank: bool) -> tuple[bool, bool]: + if example and blank: + raise ValueError("Choose either --example or --blank, not both.") + + if not example and not blank: + example = True + + if blank: + example = False + + return example, blank + + +def _validate_init_targets( + directories: list[Path], + files: list[tuple[Path, str]], + *, + workspace: Path, +) -> None: + for directory in directories: + ensure_safe_write_target(directory, workspace) + for file_path, _content in files: + ensure_safe_write_target(file_path, workspace) + ensure_safe_write_target(SKILL_FILE_PATH, workspace) + ensure_safe_write_target(SKILL_HASH_PATH, workspace) + + +def run_specleft_init( + *, + example: bool = False, + blank: bool = False, + dry_run: bool = False, +) -> dict[str, Any]: + """Run SpecLeft initialisation in MCP-safe mode.""" + try: + use_example, _ = _normalize_options(example=example, blank=blank) + except ValueError as exc: + return { + "success": False, + "error": str(exc), + "health": { + "python_version": None, + "dependencies_ok": False, + "plugin_registered": False, + }, + "created": [], + "skill_file": str(SKILL_FILE_PATH), + "next_steps": "Fix the input arguments and retry.", + } + + checks = _build_doctor_checks(verify_skill=False) + doctor_output = _build_doctor_output(checks) + health = _health_payload(checks) + + if not bool(doctor_output.get("healthy", False)): + return { + "success": False, + "error": "Environment health checks failed.", + "health": health, + "created": [], + "skill_file": str(SKILL_FILE_PATH), + "next_steps": "Run `specleft doctor` and resolve reported issues.", + "errors": doctor_output.get("errors", []), + } + + directories, files = _init_plan(example=use_example) + workspace = Path.cwd().resolve() + + try: + _validate_init_targets(directories, files, workspace=workspace) + except SecurityError as exc: + return { + "success": False, + "error": str(exc), + "health": health, + "created": [], + "skill_file": str(SKILL_FILE_PATH), + "next_steps": "Review filesystem safety and retry initialisation.", + } + + if dry_run: + planned = [str(path) for path, _content in files] + planned.extend([str(SKILL_FILE_PATH), str(SKILL_HASH_PATH)]) + planned.extend(str(directory) for directory in directories) + return { + "success": True, + "dry_run": True, + "health": health, + "created": sorted(set(planned)), + "skill_file": str(SKILL_FILE_PATH), + "next_steps": "Run specleft_init without dry_run to write files.", + } + + created_paths = [str(path) for path in _apply_init_plan(directories, files)] + skill_sync = sync_skill_files(overwrite_existing=False) + created_paths.extend(skill_sync.created) + + return { + "success": True, + "health": health, + "created": created_paths, + "skill_file": str(SKILL_FILE_PATH), + "next_steps": "Read .specleft/SKILL.md for full CLI reference", + "warnings": skill_sync.warnings, + } diff --git a/src/specleft/mcp/payloads.py b/src/specleft/mcp/payloads.py new file mode 100644 index 0000000..2005d19 --- /dev/null +++ b/src/specleft/mcp/payloads.py @@ -0,0 +1,164 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026 SpecLeft Contributors + +"""Payload builders for SpecLeft MCP resources.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from specleft.commands.contracts.payloads import build_contract_payload +from specleft.commands.status import build_status_entries, build_status_json +from specleft.utils.specs_dir import resolve_specs_dir +from specleft.validator import load_specs_directory + + +def _build_empty_status_payload(*, initialised: bool, verbose: bool) -> dict[str, Any]: + if not verbose: + return { + "initialised": initialised, + "features": 0, + "scenarios": 0, + "implemented": 0, + "skipped": 0, + "coverage_percent": 0.0, + } + + return { + "initialised": initialised, + "summary": { + "features": 0, + "scenarios": 0, + "implemented": 0, + "skipped": 0, + "coverage_percent": 0.0, + }, + "by_priority": {}, + "features": [], + } + + +def build_mcp_status_payload( + *, + verbose: bool = False, + features_dir: str | None = ".specleft/specs", + tests_dir: Path | None = Path("tests"), +) -> dict[str, Any]: + """Build the status resource payload for MCP clients.""" + resolved_features_dir = resolve_specs_dir(features_dir) + if not resolved_features_dir.exists(): + return _build_empty_status_payload(initialised=False, verbose=verbose) + + try: + config = load_specs_directory(resolved_features_dir) + except FileNotFoundError: + return _build_empty_status_payload(initialised=False, verbose=verbose) + except ValueError as exc: + if "No feature specs found" in str(exc): + return _build_empty_status_payload(initialised=True, verbose=verbose) + return _build_empty_status_payload(initialised=False, verbose=verbose) + + entries = build_status_entries(config, tests_dir or Path("tests")) + status_payload = build_status_json( + entries, + include_execution_time=False, + verbose=verbose, + ) + if isinstance(status_payload, dict): + return status_payload + return _build_empty_status_payload(initialised=True, verbose=verbose) + + +def build_mcp_contract_payload() -> dict[str, Any]: + """Build the contract resource payload for MCP clients.""" + payload = build_contract_payload() + raw_guarantees = payload.get("guarantees") + guarantees = dict(raw_guarantees) if isinstance(raw_guarantees, dict) else {} + raw_cli_api = guarantees.get("cli_api") + cli_api = dict(raw_cli_api) if isinstance(raw_cli_api, dict) else {} + raw_exit_codes = cli_api.get("exit_codes") + if isinstance(raw_exit_codes, dict): + exit_codes = dict(raw_exit_codes) + else: + exit_codes = { + "success": 0, + "error": 1, + "cancelled": 2, + } + guarantees["exit_codes"] = exit_codes + + mcp_payload = dict(payload) + mcp_payload["guarantees"] = guarantees + mcp_payload.pop("docs", None) + return mcp_payload + + +def build_mcp_guide_payload() -> dict[str, object]: + """Build the workflow guide payload for MCP clients.""" + return { + "workflow": { + "bulk_setup": [ + { + "step": 1, + "command": "plan --analyze", + "description": "Understand how the current plan aligns with the prd-template.yml", + }, + { + "step": 2, + "action": "modify", + "description": "If required, modify .specleft/templates/prd-template.yml to align PRD parsing.", + }, + { + "step": 3, + "command": "plan", + "description": "Generate features from the calibrated PRD.", + }, + { + "step": 4, + "command": "features add-scenario --add-test skeleton", + "description": "Append scenarios and scaffold tests.", + }, + ], + "incremental_setup": [ + { + "step": 1, + "command": "features add", + "description": "Add an individual feature spec.", + }, + { + "step": 2, + "command": "features add-scenario --add-test skeleton", + "description": "Append scenarios and scaffold tests.", + }, + ], + "implementation": [ + { + "step": 1, + "command": "next --limit 1", + "description": "Pick the next scenario to implement.", + }, + { + "step": 2, + "action": "implement", + "description": "Write the test logic first, then application code.", + }, + { + "step": 3, + "action": "pytest", + "description": "Run tests, fix failures, and repeat.", + }, + { + "step": 4, + "command": "features validate --strict", + "description": "Validate all specs before commit.", + }, + { + "step": 5, + "command": "coverage --threshold 100", + "description": "Verify feature coverage target.", + }, + ], + }, + "skill_file": "Run specleft_init to generate .specleft/SKILL.md with full CLI reference", + } diff --git a/src/specleft/mcp/server.py b/src/specleft/mcp/server.py new file mode 100644 index 0000000..8da0eef --- /dev/null +++ b/src/specleft/mcp/server.py @@ -0,0 +1,99 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026 SpecLeft Contributors + +"""FastMCP server wiring for SpecLeft.""" + +from __future__ import annotations + +import importlib +from typing import Any + +from specleft.commands.constants import CLI_VERSION +from specleft.mcp.init_tool import run_specleft_init +from specleft.mcp.payloads import ( + build_mcp_contract_payload, + build_mcp_guide_payload, + build_mcp_status_payload, +) + + +def _require_fastmcp() -> tuple[Any, Any]: + try: + fastmcp_module = importlib.import_module("fastmcp") + resources_module = importlib.import_module("fastmcp.resources") + except ModuleNotFoundError as exc: # pragma: no cover - exercised in CLI error path + raise RuntimeError( + "FastMCP is not installed. Install with `pip install specleft[mcp]`." + ) from exc + + FastMCP = fastmcp_module.FastMCP + FunctionResource = resources_module.FunctionResource + return FastMCP, FunctionResource + + +def build_mcp_server() -> Any: + """Create the SpecLeft MCP server instance.""" + FastMCP, FunctionResource = _require_fastmcp() + + mcp = FastMCP( + name="SpecLeft", + version=CLI_VERSION, + website_url="https://specleft.dev", + instructions=( + "Use contract, guide, and status resources before mutating project files. " + "Use specleft_init only when status reports initialised=false." + ), + ) + + mcp.add_resource( + FunctionResource( + uri="specleft://contract", + name="SpecLeft Agent Contract", + description="Safety and determinism guarantees for this SpecLeft installation.", + mime_type="application/json", + fn=build_mcp_contract_payload, + ) + ) + + mcp.add_resource( + FunctionResource( + uri="specleft://guide", + name="SpecLeft Workflow Guide", + description="Workflow and Skill guidance for agents using SpecLeft.", + mime_type="application/json", + fn=build_mcp_guide_payload, + ) + ) + + mcp.add_resource( + FunctionResource( + uri="specleft://status", + name="SpecLeft Project Status", + description="Project-level implementation and coverage summary.", + mime_type="application/json", + fn=build_mcp_status_payload, + ) + ) + + def specleft_init( + example: bool = False, + blank: bool = False, + dry_run: bool = False, + ) -> dict[str, Any]: + return run_specleft_init(example=example, blank=blank, dry_run=dry_run) + + mcp.tool( + specleft_init, + name="specleft_init", + description=( + "Initialise a SpecLeft project, run health checks, and generate .specleft/SKILL.md." + ), + ) + + return mcp + + +def run_mcp_server() -> None: + """Run the SpecLeft MCP server over stdio.""" + server = build_mcp_server() + server.run(transport="stdio", show_banner=False) diff --git a/tests/cli/test_cli_base.py b/tests/cli/test_cli_base.py index acff45c..526745e 100644 --- a/tests/cli/test_cli_base.py +++ b/tests/cli/test_cli_base.py @@ -28,3 +28,4 @@ def test_cli_help(self) -> None: assert "features" in result.output assert "contract" in result.output assert "skill" in result.output + assert "mcp" in result.output diff --git a/tests/commands/test_status.py b/tests/commands/test_status.py index 3d91447..c44909b 100644 --- a/tests/commands/test_status.py +++ b/tests/commands/test_status.py @@ -30,6 +30,21 @@ def test_status_json_includes_execution_time(self) -> None: scenarios = payload["features"][0]["scenarios"] assert scenarios[0]["execution_time"] == "slow" + def test_status_summary_json_includes_initialised(self) -> None: + runner = CliRunner() + with runner.isolated_filesystem(): + create_feature_specs( + Path("."), + feature_id="auth", + story_id="login", + scenario_id="login-success", + ) + result = runner.invoke(cli, ["status", "--format", "json"]) + assert result.exit_code == 0 + payload = json.loads(result.output) + assert payload["initialised"] is True + assert payload["features"] == 1 + def test_status_groups_by_feature_file(self) -> None: runner = CliRunner() with runner.isolated_filesystem(): diff --git a/tests/mcp/__init__.py b/tests/mcp/__init__.py new file mode 100644 index 0000000..fbb65c4 --- /dev/null +++ b/tests/mcp/__init__.py @@ -0,0 +1 @@ +"""MCP test package.""" diff --git a/tests/mcp/test_security.py b/tests/mcp/test_security.py new file mode 100644 index 0000000..524ed87 --- /dev/null +++ b/tests/mcp/test_security.py @@ -0,0 +1,61 @@ +"""Security tests for MCP init workflow.""" + +from __future__ import annotations + +import hashlib +import stat +from pathlib import Path + +import pytest + +from specleft.mcp.init_tool import ( + SecurityError, + ensure_safe_write_target, + run_specleft_init, +) +from specleft.utils.skill_integrity import verify_skill_integrity + + +def test_ensure_safe_write_target_rejects_traversal(tmp_path: Path) -> None: + with pytest.raises(SecurityError, match="Path traversal"): + ensure_safe_write_target(Path("../outside.txt"), workspace=tmp_path) + + +def test_init_rejects_symlink_targets( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.chdir(tmp_path) + outside = tmp_path.parent / "outside" + outside.mkdir(exist_ok=True) + + (tmp_path / ".specleft").mkdir(parents=True) + (tmp_path / ".specleft" / "specs").symlink_to(outside) + + payload = run_specleft_init(blank=True) + + assert payload["success"] is False + assert "symlink" in str(payload["error"]).lower() + + +def test_init_generates_verified_read_only_skill_file( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.chdir(tmp_path) + + payload = run_specleft_init(blank=True) + assert payload["success"] is True + + skill_file = tmp_path / ".specleft" / "SKILL.md" + checksum_file = tmp_path / ".specleft" / "SKILL.md.sha256" + + skill_hash = hashlib.sha256(skill_file.read_bytes()).hexdigest() + assert checksum_file.read_text().strip() == skill_hash + + mode = stat.S_IMODE(skill_file.stat().st_mode) + assert mode == 0o444 + + integrity = verify_skill_integrity().to_payload() + assert integrity["commands_simple"] is True + assert integrity["integrity"] in {"pass", "outdated"} diff --git a/tests/mcp/test_server.py b/tests/mcp/test_server.py new file mode 100644 index 0000000..8096b60 --- /dev/null +++ b/tests/mcp/test_server.py @@ -0,0 +1,181 @@ +"""Integration tests for the SpecLeft MCP server.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest + +from specleft.commands.status import build_status_entries, build_status_json +from specleft import specleft +from specleft.mcp.payloads import build_mcp_status_payload +from specleft.mcp.server import build_mcp_server +from specleft.validator import load_specs_directory +from tests.helpers.specs import create_feature_specs + + +@pytest.fixture +def mcp_client() -> Any: + """Return an in-memory FastMCP client for the SpecLeft server.""" + fastmcp = pytest.importorskip("fastmcp") + return fastmcp.Client(build_mcp_server()) + + +def _resource_json(result: list[Any]) -> dict[str, object]: + text = result[0].text + return json.loads(text) + + +@pytest.mark.asyncio +async def test_server_lists_three_resources(mcp_client: Any) -> None: + async with mcp_client: + resources = await mcp_client.list_resources() + + uris = {str(resource.uri) for resource in resources} + assert uris == { + "specleft://contract", + "specleft://guide", + "specleft://status", + } + + +@pytest.mark.asyncio +async def test_server_lists_one_tool(mcp_client: Any) -> None: + async with mcp_client: + tools = await mcp_client.list_tools() + + assert len(tools) == 1 + assert tools[0].name == "specleft_init" + + +@pytest.mark.asyncio +async def test_contract_and_guide_resources_are_json(mcp_client: Any) -> None: + async with mcp_client: + contract_result = await mcp_client.read_resource("specleft://contract") + guide_result = await mcp_client.read_resource("specleft://guide") + + contract_payload = _resource_json(contract_result) + guide_payload = _resource_json(guide_result) + + assert contract_payload["contract_version"] + assert "guarantees" in contract_payload + assert guide_payload["workflow"] + assert "skill_file" in guide_payload + + +@pytest.mark.asyncio +@specleft( + feature_id="feature-mcp-server", + scenario_id="agent-discovery-flow-uses-resources-and-init-tool", +) +async def test_agent_discovery_flow_uses_resources_and_init_tool( + mcp_client: Any, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.chdir(tmp_path) + + with specleft.step("Given an agent connects to the SpecLeft MCP server"): + pass + + with specleft.step("When the agent reads contract, guide, and status resources"): + async with mcp_client: + contract_result = await mcp_client.read_resource("specleft://contract") + guide_result = await mcp_client.read_resource("specleft://guide") + status_result = await mcp_client.read_resource("specleft://status") + + contract_payload = _resource_json(contract_result) + guide_payload = _resource_json(guide_result) + status_payload = _resource_json(status_result) + + with specleft.step("Then status reports initialised=false before setup"): + assert "guarantees" in contract_payload + assert "workflow" in guide_payload + assert status_payload["initialised"] is False + + with specleft.step("And the agent calls specleft_init to bootstrap the project"): + async with mcp_client: + init_result = await mcp_client.call_tool("specleft_init", {"example": True}) + init_payload = json.loads(init_result.content[0].text) + + with specleft.step("Then initialisation succeeds and creates the skill file"): + assert init_payload["success"] is True + assert (tmp_path / ".specleft" / "SKILL.md").exists() + + with specleft.step("And status reflects initialised project state"): + async with mcp_client: + status_after = await mcp_client.read_resource("specleft://status") + status_after_payload = _resource_json(status_after) + assert status_after_payload["initialised"] is True + assert status_after_payload["features"] >= 1 + + +@pytest.mark.asyncio +async def test_init_tool_dry_run_writes_nothing( + mcp_client: Any, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.chdir(tmp_path) + before = {path.relative_to(tmp_path) for path in tmp_path.rglob("*")} + + async with mcp_client: + result = await mcp_client.call_tool("specleft_init", {"dry_run": True}) + payload = json.loads(result.content[0].text) + + after = {path.relative_to(tmp_path) for path in tmp_path.rglob("*")} + + assert payload["success"] is True + assert payload["dry_run"] is True + assert before == after + + +@pytest.mark.asyncio +async def test_init_tool_is_idempotent( + mcp_client: Any, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.chdir(tmp_path) + + async with mcp_client: + first = await mcp_client.call_tool("specleft_init", {"blank": True}) + second = await mcp_client.call_tool("specleft_init", {"blank": True}) + + first_payload = json.loads(first.content[0].text) + second_payload = json.loads(second.content[0].text) + + assert first_payload["success"] is True + assert second_payload["success"] is True + + +def test_status_payload_verbose_shape( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.chdir(tmp_path) + create_feature_specs( + tmp_path, + feature_id="feature-auth", + story_id="login", + scenario_id="user-can-login", + ) + + payload = build_mcp_status_payload(verbose=True) + config = load_specs_directory(Path(".specleft/specs")) + entries = build_status_entries(config, Path("tests")) + expected = build_status_json( + entries, + include_execution_time=False, + verbose=True, + ) + assert isinstance(expected, dict) + + payload_without_timestamp = dict(payload) + expected_without_timestamp = dict(expected) + payload_without_timestamp.pop("timestamp", None) + expected_without_timestamp.pop("timestamp", None) + + assert payload_without_timestamp == expected_without_timestamp diff --git a/tests/mcp/test_token_budget.py b/tests/mcp/test_token_budget.py new file mode 100644 index 0000000..b24ce56 --- /dev/null +++ b/tests/mcp/test_token_budget.py @@ -0,0 +1,68 @@ +"""Token usage tests for MCP payloads and declarations.""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from typing import Any + +import pytest + +from specleft.mcp.payloads import build_mcp_contract_payload, build_mcp_guide_payload +from specleft.mcp.server import build_mcp_server + + +def _count_tokens(payload: str) -> int: + tiktoken = pytest.importorskip("tiktoken") + try: + encoding = tiktoken.get_encoding("cl100k_base") + except Exception as exc: # pragma: no cover - depends on network/cache state + pytest.skip(f"Unable to load cl100k_base encoding: {exc}") + return len(encoding.encode(payload)) + + +def _compact_json(data: Any) -> str: + return json.dumps(data, separators=(",", ":"), sort_keys=True) + + +def test_contract_payload_within_budget() -> None: + tokens = _count_tokens(_compact_json(build_mcp_contract_payload())) + assert tokens <= 170 + + +def test_guide_payload_within_budget() -> None: + tokens = _count_tokens(_compact_json(build_mcp_guide_payload())) + assert tokens <= 320 + + +@pytest.mark.asyncio +async def test_declarations_within_budget() -> None: + fastmcp = pytest.importorskip("fastmcp") + client_factory: Callable[[Any], Any] = fastmcp.Client + client = client_factory(build_mcp_server()) + + async with client: + resources = await client.list_resources() + tools = await client.list_tools() + + declaration_payload = { + "resources": [ + { + "uri": str(resource.uri), + "name": resource.name, + "description": resource.description, + } + for resource in resources + ], + "tools": [ + { + "name": tool.name, + "description": tool.description, + "input_schema": tool.inputSchema, + } + for tool in tools + ], + } + + tokens = _count_tokens(_compact_json(declaration_payload)) + assert tokens <= 220