thaloco/pyro-mcp
1
Fork 0
Code Pull requests Releases Packages Activity Actions

Add MCP tool profiles for workspace chat flows

Browse source 
Expose stable MCP/server tool profiles so chat hosts can start narrow and widen only when needed. This adds vm-run, workspace-core, and workspace-full across the CLI serve path, Pyro.create_server(), and the package-level create_server() factory while keeping workspace-full as the default.

Register profile-specific tool sets from one shared contract mapping, and narrow the workspace-core schemas so secrets, network policy, shells, services, snapshots, and disk tools do not leak into the default persistent chat profile. The full surface remains available unchanged under workspace-full.

Refresh the public docs and examples around the profile progression, add a canonical OpenAI Responses workspace-core example, mark the 3.4.0 roadmap milestone done, and verify with uv lock, UV_CACHE_DIR=.uv-cache make check, UV_CACHE_DIR=.uv-cache make dist-check, and a real guest-backed workspace-core smoke for create, file write, exec, diff, export, reset, and delete.
This commit is contained in:
Thales Maciel 2026-03-12 23:52:13 -03:00
parent 446f7fce04
commit eecfd7a7d7
23 changed files with 984 additions and 511 deletions
Download patch file Download diff file Expand all files Collapse all files

84
tests/test_api.py
View file

@ -6,6 +6,11 @@ from pathlib import Path
from typing import Any, cast
from pyro_mcp.api import Pyro
from pyro_mcp.contract import (
PUBLIC_MCP_VM_RUN_PROFILE_TOOLS,
PUBLIC_MCP_WORKSPACE_CORE_PROFILE_TOOLS,
PUBLIC_MCP_WORKSPACE_FULL_PROFILE_TOOLS,
)
from pyro_mcp.vm_manager import VmManager
from pyro_mcp.vm_network import TapNetworkManager
@ -47,37 +52,54 @@ def test_pyro_create_server_registers_vm_run(tmp_path: Path) -> None:
return sorted(tool.name for tool in tools)
tool_names = asyncio.run(_run())
assert "vm_run" in tool_names
assert "vm_create" in tool_names
assert "workspace_create" in tool_names
assert "workspace_list" in tool_names
assert "workspace_update" in tool_names
assert "workspace_start" in tool_names
assert "workspace_stop" in tool_names
assert "workspace_diff" in tool_names
assert "workspace_sync_push" in tool_names
assert "workspace_export" in tool_names
assert "workspace_file_list" in tool_names
assert "workspace_file_read" in tool_names
assert "workspace_file_write" in tool_names
assert "workspace_patch_apply" in tool_names
assert "workspace_disk_export" in tool_names
assert "workspace_disk_list" in tool_names
assert "workspace_disk_read" in tool_names
assert "snapshot_create" in tool_names
assert "snapshot_list" in tool_names
assert "snapshot_delete" in tool_names
assert "workspace_reset" in tool_names
assert "shell_open" in tool_names
assert "shell_read" in tool_names
assert "shell_write" in tool_names
assert "shell_signal" in tool_names
assert "shell_close" in tool_names
assert "service_start" in tool_names
assert "service_list" in tool_names
assert "service_status" in tool_names
assert "service_logs" in tool_names
assert "service_stop" in tool_names
assert tuple(tool_names) == tuple(sorted(PUBLIC_MCP_WORKSPACE_FULL_PROFILE_TOOLS))
def test_pyro_create_server_vm_run_profile_registers_only_vm_run(tmp_path: Path) -> None:
pyro = Pyro(
manager=VmManager(
backend_name="mock",
base_dir=tmp_path / "vms",
network_manager=TapNetworkManager(enabled=False),
)
)
async def _run() -> list[str]:
server = pyro.create_server(profile="vm-run")
tools = await server.list_tools()
return sorted(tool.name for tool in tools)
assert tuple(asyncio.run(_run())) == PUBLIC_MCP_VM_RUN_PROFILE_TOOLS
def test_pyro_create_server_workspace_core_profile_registers_expected_tools_and_schemas(
tmp_path: Path,
) -> None:
pyro = Pyro(
manager=VmManager(
backend_name="mock",
base_dir=tmp_path / "vms",
network_manager=TapNetworkManager(enabled=False),
)
)
async def _run() -> tuple[list[str], dict[str, dict[str, Any]]]:
server = pyro.create_server(profile="workspace-core")
tools = await server.list_tools()
tool_map = {tool.name: tool.model_dump() for tool in tools}
return sorted(tool_map), tool_map
tool_names, tool_map = asyncio.run(_run())
assert tuple(tool_names) == tuple(sorted(PUBLIC_MCP_WORKSPACE_CORE_PROFILE_TOOLS))
create_properties = tool_map["workspace_create"]["inputSchema"]["properties"]
assert "network_policy" not in create_properties
assert "secrets" not in create_properties
exec_properties = tool_map["workspace_exec"]["inputSchema"]["properties"]
assert "secret_env" not in exec_properties
assert "shell_open" not in tool_map
assert "service_start" not in tool_map
assert "snapshot_create" not in tool_map
assert "workspace_disk_export" not in tool_map
def test_pyro_vm_run_tool_executes(tmp_path: Path) -> None:

11
tests/test_cli.py
View file

@ -63,6 +63,10 @@ def test_cli_subcommand_help_includes_examples_and_guidance() -> None:
mcp_help = _subparser_choice(_subparser_choice(parser, "mcp"), "serve").format_help()
assert "Expose pyro tools over stdio for an MCP client." in mcp_help
assert "--profile" in mcp_help
assert "workspace-core" in mcp_help
assert "workspace-full" in mcp_help
assert "vm-run" in mcp_help
assert "Use this from an MCP client config after the CLI evaluation path works." in mcp_help
workspace_help = _subparser_choice(parser, "workspace").format_help()
@ -3435,7 +3439,8 @@ def test_cli_mcp_runs_stdio_transport(monkeypatch: pytest.MonkeyPatch) -> None:
observed: dict[str, str] = {}
class StubPyro:
def create_server(self) -> Any:
def create_server(self, *, profile: str) -> Any:
observed["profile"] = profile
return type(
"StubServer",
(),
@ -3444,12 +3449,12 @@ def test_cli_mcp_runs_stdio_transport(monkeypatch: pytest.MonkeyPatch) -> None:
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(command="mcp", mcp_command="serve")
return argparse.Namespace(command="mcp", mcp_command="serve", profile="workspace-core")
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
assert observed == {"transport": "stdio"}
assert observed == {"profile": "workspace-core", "transport": "stdio"}
def test_cli_demo_default_prints_json(

11
tests/test_public_contract.py
View file

@ -16,6 +16,8 @@ from pyro_mcp.contract import (
PUBLIC_CLI_COMMANDS,
PUBLIC_CLI_DEMO_SUBCOMMANDS,
PUBLIC_CLI_ENV_SUBCOMMANDS,
PUBLIC_CLI_MCP_SERVE_FLAGS,
PUBLIC_CLI_MCP_SUBCOMMANDS,
PUBLIC_CLI_RUN_FLAGS,
PUBLIC_CLI_WORKSPACE_CREATE_FLAGS,
PUBLIC_CLI_WORKSPACE_DIFF_FLAGS,
@ -54,6 +56,7 @@ from pyro_mcp.contract import (
PUBLIC_CLI_WORKSPACE_SYNC_PUSH_FLAGS,
PUBLIC_CLI_WORKSPACE_SYNC_SUBCOMMANDS,
PUBLIC_CLI_WORKSPACE_UPDATE_FLAGS,
PUBLIC_MCP_PROFILES,
PUBLIC_MCP_TOOLS,
PUBLIC_SDK_METHODS,
)
@ -99,6 +102,14 @@ def test_public_cli_help_lists_commands_and_run_flags() -> None:
env_help_text = _subparser_choice(parser, "env").format_help()
for subcommand_name in PUBLIC_CLI_ENV_SUBCOMMANDS:
assert subcommand_name in env_help_text
mcp_help_text = _subparser_choice(parser, "mcp").format_help()
for subcommand_name in PUBLIC_CLI_MCP_SUBCOMMANDS:
assert subcommand_name in mcp_help_text
mcp_serve_help_text = _subparser_choice(_subparser_choice(parser, "mcp"), "serve").format_help()
for flag in PUBLIC_CLI_MCP_SERVE_FLAGS:
assert flag in mcp_serve_help_text
for profile_name in PUBLIC_MCP_PROFILES:
assert profile_name in mcp_serve_help_text
workspace_help_text = _subparser_choice(parser, "workspace").format_help()
for subcommand_name in PUBLIC_CLI_WORKSPACE_SUBCOMMANDS:

157
tests/test_server.py
View file

@ -7,6 +7,11 @@ from typing import Any, cast
import pytest
import pyro_mcp.server as server_module
from pyro_mcp.contract import (
PUBLIC_MCP_VM_RUN_PROFILE_TOOLS,
PUBLIC_MCP_WORKSPACE_CORE_PROFILE_TOOLS,
PUBLIC_MCP_WORKSPACE_FULL_PROFILE_TOOLS,
)
from pyro_mcp.server import create_server
from pyro_mcp.vm_manager import VmManager
from pyro_mcp.vm_network import TapNetworkManager
@ -25,42 +30,37 @@ def test_create_server_registers_vm_tools(tmp_path: Path) -> None:
return sorted(tool.name for tool in tools)
tool_names = asyncio.run(_run())
assert "vm_create" in tool_names
assert "vm_exec" in tool_names
assert "vm_list_environments" in tool_names
assert "vm_network_info" in tool_names
assert "vm_run" in tool_names
assert "vm_status" in tool_names
assert "workspace_create" in tool_names
assert "workspace_list" in tool_names
assert "workspace_update" in tool_names
assert "workspace_start" in tool_names
assert "workspace_stop" in tool_names
assert "workspace_diff" in tool_names
assert "workspace_export" in tool_names
assert "workspace_file_list" in tool_names
assert "workspace_file_read" in tool_names
assert "workspace_file_write" in tool_names
assert "workspace_patch_apply" in tool_names
assert "workspace_disk_export" in tool_names
assert "workspace_disk_list" in tool_names
assert "workspace_disk_read" in tool_names
assert "workspace_logs" in tool_names
assert "workspace_sync_push" in tool_names
assert "shell_open" in tool_names
assert "shell_read" in tool_names
assert "shell_write" in tool_names
assert "shell_signal" in tool_names
assert "shell_close" in tool_names
assert "service_start" in tool_names
assert "service_list" in tool_names
assert "service_status" in tool_names
assert "service_logs" in tool_names
assert "service_stop" in tool_names
assert "snapshot_create" in tool_names
assert "snapshot_delete" in tool_names
assert "snapshot_list" in tool_names
assert "workspace_reset" in tool_names
assert tuple(tool_names) == tuple(sorted(PUBLIC_MCP_WORKSPACE_FULL_PROFILE_TOOLS))
def test_create_server_vm_run_profile_registers_only_vm_run(tmp_path: Path) -> None:
manager = VmManager(
backend_name="mock",
base_dir=tmp_path / "vms",
network_manager=TapNetworkManager(enabled=False),
)
async def _run() -> list[str]:
server = create_server(manager=manager, profile="vm-run")
tools = await server.list_tools()
return sorted(tool.name for tool in tools)
assert tuple(asyncio.run(_run())) == PUBLIC_MCP_VM_RUN_PROFILE_TOOLS
def test_create_server_workspace_core_profile_registers_expected_tools(tmp_path: Path) -> None:
manager = VmManager(
backend_name="mock",
base_dir=tmp_path / "vms",
network_manager=TapNetworkManager(enabled=False),
)
async def _run() -> list[str]:
server = create_server(manager=manager, profile="workspace-core")
tools = await server.list_tools()
return sorted(tool.name for tool in tools)
assert tuple(asyncio.run(_run())) == tuple(sorted(PUBLIC_MCP_WORKSPACE_CORE_PROFILE_TOOLS))
def test_vm_run_round_trip(tmp_path: Path) -> None:
@ -193,6 +193,91 @@ def test_server_main_runs_stdio_transport(monkeypatch: pytest.MonkeyPatch) -> No
assert called == {"transport": "stdio"}
def test_workspace_core_profile_round_trip(tmp_path: Path) -> None:
manager = VmManager(
backend_name="mock",
base_dir=tmp_path / "vms",
network_manager=TapNetworkManager(enabled=False),
)
source_dir = tmp_path / "seed"
source_dir.mkdir()
(source_dir / "note.txt").write_text("old\n", encoding="utf-8")
def _extract_structured(raw_result: object) -> dict[str, Any]:
if not isinstance(raw_result, tuple) or len(raw_result) != 2:
raise TypeError("unexpected call_tool result shape")
_, structured = raw_result
if not isinstance(structured, dict):
raise TypeError("expected structured dictionary result")
return cast(dict[str, Any], structured)
async def _run() -> tuple[dict[str, Any], ...]:
server = create_server(manager=manager, profile="workspace-core")
created = _extract_structured(
await server.call_tool(
"workspace_create",
{
"environment": "debian:12-base",
"allow_host_compat": True,
"seed_path": str(source_dir),
"name": "chat-loop",
"labels": {"issue": "123"},
},
)
)
workspace_id = str(created["workspace_id"])
written = _extract_structured(
await server.call_tool(
"workspace_file_write",
{
"workspace_id": workspace_id,
"path": "note.txt",
"text": "fixed\n",
},
)
)
executed = _extract_structured(
await server.call_tool(
"workspace_exec",
{
"workspace_id": workspace_id,
"command": "cat note.txt",
},
)
)
diffed = _extract_structured(
await server.call_tool("workspace_diff", {"workspace_id": workspace_id})
)
export_path = tmp_path / "exported-note.txt"
exported = _extract_structured(
await server.call_tool(
"workspace_export",
{
"workspace_id": workspace_id,
"path": "note.txt",
"output_path": str(export_path),
},
)
)
reset = _extract_structured(
await server.call_tool("workspace_reset", {"workspace_id": workspace_id})
)
deleted = _extract_structured(
await server.call_tool("workspace_delete", {"workspace_id": workspace_id})
)
return created, written, executed, diffed, exported, reset, deleted
created, written, executed, diffed, exported, reset, deleted = asyncio.run(_run())
assert created["name"] == "chat-loop"
assert created["labels"] == {"issue": "123"}
assert written["bytes_written"] == len("fixed\n".encode("utf-8"))
assert executed["stdout"] == "fixed\n"
assert diffed["changed"] is True
assert Path(str(exported["output_path"])).read_text(encoding="utf-8") == "fixed\n"
assert reset["command_count"] == 0
assert deleted["deleted"] is True
def test_workspace_tools_round_trip(tmp_path: Path) -> None:
manager = VmManager(
backend_name="mock",