Add Python MCP server option
This commit is contained in:
25
README.md
25
README.md
@@ -68,6 +68,31 @@ Smoke test:
|
||||
npm run mcp:smoke
|
||||
```
|
||||
|
||||
## Python MCP Server
|
||||
|
||||
If you prefer Python over Node for the MCP wrapper, the repository also ships a
|
||||
Python FastMCP variant.
|
||||
|
||||
Setup:
|
||||
|
||||
```bash
|
||||
python3 -m venv .venv-mcp
|
||||
. .venv-mcp/bin/activate
|
||||
pip install -r requirements-mcp.txt
|
||||
```
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
python mcp_server.py
|
||||
```
|
||||
|
||||
Python smoke test:
|
||||
|
||||
```bash
|
||||
python scripts/mcp_smoke_test.py
|
||||
```
|
||||
|
||||
## Build
|
||||
|
||||
```bash
|
||||
|
||||
25
docs/mcp.md
25
docs/mcp.md
@@ -93,6 +93,31 @@ That test:
|
||||
- lists tools
|
||||
- calls `pack_binary_info`
|
||||
|
||||
## Python variant
|
||||
|
||||
The repo also contains a Python FastMCP server for teams that prefer Python for
|
||||
agent tooling.
|
||||
|
||||
Setup:
|
||||
|
||||
```bash
|
||||
python3 -m venv .venv-mcp
|
||||
. .venv-mcp/bin/activate
|
||||
pip install -r requirements-mcp.txt
|
||||
```
|
||||
|
||||
Run:
|
||||
|
||||
```bash
|
||||
python mcp_server.py
|
||||
```
|
||||
|
||||
Smoke test:
|
||||
|
||||
```bash
|
||||
python scripts/mcp_smoke_test.py
|
||||
```
|
||||
|
||||
## Claude Desktop style config example
|
||||
|
||||
```json
|
||||
|
||||
165
mcp_server.py
Normal file
165
mcp_server.py
Normal file
@@ -0,0 +1,165 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent
|
||||
DEFAULT_BINARY = REPO_ROOT / "build" / "m2pack"
|
||||
ENV_BINARY = "M2PACK_BINARY"
|
||||
|
||||
mcp = FastMCP("m2pack-secure")
|
||||
|
||||
|
||||
def _resolve_binary() -> Path:
|
||||
env_value = os.environ.get(ENV_BINARY)
|
||||
if env_value:
|
||||
candidate = Path(env_value).expanduser()
|
||||
if candidate.is_file():
|
||||
return candidate
|
||||
|
||||
if DEFAULT_BINARY.is_file():
|
||||
return DEFAULT_BINARY
|
||||
|
||||
which = shutil.which("m2pack")
|
||||
if which:
|
||||
return Path(which)
|
||||
|
||||
raise FileNotFoundError(
|
||||
f"m2pack binary not found. Build {DEFAULT_BINARY} or set {ENV_BINARY}."
|
||||
)
|
||||
|
||||
|
||||
def _run_cli(*args: str) -> dict[str, Any]:
|
||||
binary = _resolve_binary()
|
||||
cmd = [str(binary), *args, "--json"]
|
||||
proc = subprocess.run(
|
||||
cmd,
|
||||
cwd=REPO_ROOT,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
if proc.returncode != 0:
|
||||
detail = proc.stderr.strip() or proc.stdout.strip() or f"exit code {proc.returncode}"
|
||||
raise RuntimeError(f"m2pack failed: {detail}")
|
||||
|
||||
stdout = proc.stdout.strip()
|
||||
if not stdout:
|
||||
return {"ok": True}
|
||||
|
||||
try:
|
||||
return json.loads(stdout)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise RuntimeError(f"m2pack returned non-JSON output: {stdout}") from exc
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def pack_keygen(out_dir: str) -> dict[str, Any]:
|
||||
"""Generate a new master content key and Ed25519 signing keypair."""
|
||||
return _run_cli("keygen", "--out-dir", out_dir)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def pack_build(
|
||||
input_dir: str,
|
||||
output_archive: str,
|
||||
key_file: str,
|
||||
signing_secret_key_file: str,
|
||||
) -> dict[str, Any]:
|
||||
"""Build an .m2p archive from a source directory."""
|
||||
return _run_cli(
|
||||
"build",
|
||||
"--input",
|
||||
input_dir,
|
||||
"--output",
|
||||
output_archive,
|
||||
"--key",
|
||||
key_file,
|
||||
"--sign-secret-key",
|
||||
signing_secret_key_file,
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def pack_list(archive: str) -> dict[str, Any]:
|
||||
"""List entries in an .m2p archive."""
|
||||
return _run_cli("list", "--archive", archive)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def pack_verify(
|
||||
archive: str,
|
||||
public_key_file: str | None = None,
|
||||
key_file: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Verify archive manifest integrity and optionally decrypt all entries."""
|
||||
args = ["verify", "--archive", archive]
|
||||
if public_key_file:
|
||||
args.extend(["--public-key", public_key_file])
|
||||
if key_file:
|
||||
args.extend(["--key", key_file])
|
||||
return _run_cli(*args)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def pack_extract(
|
||||
archive: str,
|
||||
output_dir: str,
|
||||
key_file: str,
|
||||
) -> dict[str, Any]:
|
||||
"""Extract an .m2p archive into a directory."""
|
||||
return _run_cli(
|
||||
"extract",
|
||||
"--archive",
|
||||
archive,
|
||||
"--output",
|
||||
output_dir,
|
||||
"--key",
|
||||
key_file,
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def pack_export_client_config(
|
||||
key_file: str,
|
||||
public_key_file: str,
|
||||
output_header: str,
|
||||
) -> dict[str, Any]:
|
||||
"""Generate M2PackKeys.h for the Windows client tree."""
|
||||
return _run_cli(
|
||||
"export-client-config",
|
||||
"--key",
|
||||
key_file,
|
||||
"--public-key",
|
||||
public_key_file,
|
||||
"--output",
|
||||
output_header,
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def pack_binary_info() -> dict[str, Any]:
|
||||
"""Report which m2pack binary the server will execute."""
|
||||
binary = _resolve_binary()
|
||||
return {
|
||||
"ok": True,
|
||||
"binary": str(binary),
|
||||
"repo_root": str(REPO_ROOT),
|
||||
}
|
||||
|
||||
|
||||
def main() -> None:
|
||||
mcp.run()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
1
requirements-mcp.txt
Normal file
1
requirements-mcp.txt
Normal file
@@ -0,0 +1 @@
|
||||
mcp==1.27.0
|
||||
40
scripts/mcp_smoke_test.py
Normal file
40
scripts/mcp_smoke_test.py
Normal file
@@ -0,0 +1,40 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import anyio
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from mcp import ClientSession, StdioServerParameters
|
||||
from mcp.client.stdio import stdio_client
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
server_params = StdioServerParameters(
|
||||
command=str(REPO_ROOT / ".venv-mcp" / "bin" / "python"),
|
||||
args=[str(REPO_ROOT / "mcp_server.py")],
|
||||
cwd=str(REPO_ROOT),
|
||||
env={
|
||||
**os.environ,
|
||||
"M2PACK_BINARY": os.environ.get("M2PACK_BINARY", str(REPO_ROOT / "build" / "m2pack")),
|
||||
},
|
||||
)
|
||||
|
||||
async with stdio_client(server_params) as (read_stream, write_stream):
|
||||
async with ClientSession(read_stream, write_stream) as session:
|
||||
await session.initialize()
|
||||
tools = await session.list_tools()
|
||||
binary_info = await session.call_tool("pack_binary_info", {})
|
||||
print({
|
||||
"ok": True,
|
||||
"tool_count": len(tools.tools),
|
||||
"tool_names": [tool.name for tool in tools.tools],
|
||||
"binary_info": getattr(binary_info, "structuredContent", None),
|
||||
})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
anyio.run(main)
|
||||
Reference in New Issue
Block a user