odysseus-mcp

#!/usr/bin/env python3
"""odysseus-mcp — shell wrapper for MCP (Model Context Protocol) servers.

MCP servers are configured in the `mcp_servers` table. The web app's
McpManager handles live connections; this CLI manages the *config*
(read, add, enable/disable, delete). Runtime connection state lives
in the manager process — query it via the web UI or `curl /api/mcp/servers`.

    odysseus-mcp list                    # configured servers
    odysseus-mcp show SERVER_ID
    odysseus-mcp enable SERVER_ID
    odysseus-mcp disable SERVER_ID
    odysseus-mcp add --name "Filesystem" --transport stdio \\
                    --command npx --args '["@modelcontextprotocol/server-filesystem","/tmp"]'
    odysseus-mcp delete SERVER_ID
"""

from __future__ import annotations
import sys
import os, sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "_lib"))
from cli import quiet_logs, emit, fail, common_parser, run, REPO_ROOT as _REPO_ROOT
quiet_logs()

import argparse, json, logging, os, sys, uuid
from pathlib import Path

try:
    from core.database import SessionLocal, McpServer
    quiet_logs()
except ModuleNotFoundError as e:
    sys.stderr.write(f"error: {e}\nhint: run from repo root with venv active.\n")
    sys.exit(2)


def _json_list(raw) -> list:
    try:
        value = json.loads(raw) if raw else []
    except (TypeError, json.JSONDecodeError):
        return []
    return value if isinstance(value, list) else []


def _json_dict(raw) -> dict:
    try:
        value = json.loads(raw) if raw else {}
    except (TypeError, json.JSONDecodeError):
        return {}
    return value if isinstance(value, dict) else {}


def _serialize(s: "McpServer", redact_env: bool = True) -> dict:
    args_arr = _json_list(s.args)
    env_obj = _json_dict(s.env)
    if redact_env and isinstance(env_obj, dict):
        env_obj = {k: ("***" if v else "") for k, v in env_obj.items()}
    return {
        "id": s.id,
        "name": s.name,
        "transport": s.transport,
        "command": s.command or "",
        "args": args_arr,
        "env": env_obj,
        "url": s.url or "",
        "is_enabled": bool(s.is_enabled),
        "has_oauth": bool(s.oauth_config),
        "created_at": s.created_at.isoformat() if s.created_at else "",
    }


def cmd_list(args):
    db = SessionLocal()
    try:
        rows = db.query(McpServer).order_by(McpServer.name.asc()).all()
        emit([_serialize(s) for s in rows], args)
    finally:
        db.close()


def cmd_show(args):
    db = SessionLocal()
    try:
        s = db.get(McpServer, args.id)
        if not s:
            fail(f"no MCP server with id {args.id!r}")
        # show shows env keys; values still redacted unless --reveal
        out = _serialize(s, redact_env=not args.reveal)
        emit(out, args)
    finally:
        db.close()


def cmd_enable(args):
    _set_enabled(args.id, True, args)


def cmd_disable(args):
    _set_enabled(args.id, False, args)


def _set_enabled(server_id: str, enabled: bool, args):
    db = SessionLocal()
    try:
        s = db.get(McpServer, server_id)
        if not s:
            fail(f"no MCP server with id {server_id!r}")
        s.is_enabled = enabled
        db.commit()
        emit({"ok": True, "id": s.id, "is_enabled": s.is_enabled}, args)
    finally:
        db.close()


def cmd_add(args):
    if args.transport == "stdio" and not args.command:
        fail("--command is required for stdio transport")
    if args.transport == "sse" and not args.url:
        fail("--url is required for sse transport")
    try:
        args_arr = json.loads(args.args) if args.args else []
        if not isinstance(args_arr, list):
            raise ValueError("--args must be a JSON array")
    except (json.JSONDecodeError, ValueError) as e:
        fail(f"invalid --args: {e}")
    try:
        env_obj = json.loads(args.env) if args.env else {}
        if not isinstance(env_obj, dict):
            raise ValueError("--env must be a JSON object")
    except (json.JSONDecodeError, ValueError) as e:
        fail(f"invalid --env: {e}")

    db = SessionLocal()
    try:
        s = McpServer(
            id=str(uuid.uuid4()),
            name=args.name,
            transport=args.transport,
            command=args.command or None,
            args=json.dumps(args_arr) if args_arr else None,
            env=json.dumps(env_obj) if env_obj else None,
            url=args.url or None,
            is_enabled=not args.disabled,
        )
        db.add(s)
        db.commit()
        db.refresh(s)
        emit(_serialize(s), args)
    finally:
        db.close()


def cmd_delete(args):
    db = SessionLocal()
    try:
        s = db.get(McpServer, args.id)
        if not s:
            fail(f"no MCP server with id {args.id!r}")
        snap = _serialize(s)
        db.delete(s)
        db.commit()
        emit({"ok": True, "deleted": snap}, args)
    finally:
        db.close()


def _build_parser():
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument("--pretty", action="store_true")
    p = argparse.ArgumentParser(prog="odysseus-mcp", parents=[common])
    sub = p.add_subparsers(dest="cmd", required=True)

    pl = sub.add_parser("list", parents=[common])
    pl.set_defaults(func=cmd_list)

    psh = sub.add_parser("show", parents=[common])
    psh.add_argument("id")
    psh.add_argument("--reveal", action="store_true", help="show env values unredacted")
    psh.set_defaults(func=cmd_show)

    pe = sub.add_parser("enable", parents=[common])
    pe.add_argument("id")
    pe.set_defaults(func=cmd_enable)

    pdis = sub.add_parser("disable", parents=[common])
    pdis.add_argument("id")
    pdis.set_defaults(func=cmd_disable)

    pa = sub.add_parser("add", parents=[common])
    pa.add_argument("--name", required=True)
    pa.add_argument("--transport", choices=["stdio", "sse"], default="stdio")
    pa.add_argument("--command", help="stdio: executable to run")
    pa.add_argument("--args", help="JSON array of args, e.g. '[\"--root\",\"/tmp\"]'")
    pa.add_argument("--env", help="JSON object of env vars, e.g. '{\"KEY\":\"value\"}'")
    pa.add_argument("--url", help="sse: server URL")
    pa.add_argument("--disabled", action="store_true", help="create disabled")
    pa.set_defaults(func=cmd_add)

    pd = sub.add_parser("delete", parents=[common])
    pd.add_argument("id")
    pd.set_defaults(func=cmd_delete)

    return p


if __name__ == "__main__":
    sys.exit(run(_build_parser()))