mcp_server.py

"""First-party MCP surface for OpenCLI Admin.

The same server is available over stdio (``opencli-mcp``) and as the built-in
``/mcp`` Streamable HTTP endpoint mounted by :mod:`backend.main`.  The official
MCP v2 SDK supplies the 2026-07-28 stateless protocol, ``server/discover``,
cache metadata, and legacy protocol translation.
"""

from __future__ import annotations

import os
from typing import Any
from urllib.parse import urlsplit

import httpx
from mcp.server import MCPServer
from mcp.server.mcpserver import Context
from mcp.server.transport_security import TransportSecuritySettings
from mcp_types import ToolAnnotations

from backend.config import get_settings

MCP_PROTOCOL_VERSION = "2026-07-28"

READ_ONLY_TOOL = ToolAnnotations(
    read_only_hint=True,
    destructive_hint=False,
    idempotent_hint=True,
    open_world_hint=True,
)
WRITE_TOOL = ToolAnnotations(
    read_only_hint=False,
    destructive_hint=False,
    idempotent_hint=False,
    open_world_hint=True,
)
IDEMPOTENT_WRITE_TOOL = ToolAnnotations(
    read_only_hint=False,
    destructive_hint=False,
    idempotent_hint=True,
    open_world_hint=True,
)


def _api_base_url() -> str:
    settings = get_settings()
    value = getattr(
        settings,
        "opencli_admin_api_url",
        os.environ.get("OPENCLI_ADMIN_API_URL", "http://localhost:8031"),
    )
    return str(value).rstrip("/")


def _auth_headers(context: Context | None = None) -> dict[str, str]:
    """Separate fleet transport access from the downstream caller identity."""

    settings = get_settings()
    headers: dict[str, str] = {}
    fleet_token = settings.api_auth_token.strip()
    if fleet_token:
        headers["X-API-Token"] = fleet_token

    caller_token = ""
    request_headers = context.headers if context is not None else None
    if request_headers is not None:
        scheme, _, credential = request_headers.get("authorization", "").partition(" ")
        if scheme.lower() == "bearer" and credential and credential != fleet_token:
            headers["Authorization"] = f"Bearer {credential}"
    else:
        caller_token = str(
            getattr(
                settings,
                "opencli_mcp_caller_token",
                os.environ.get("OPENCLI_MCP_CALLER_TOKEN", ""),
            )
        ).strip()
        if caller_token:
            headers["Authorization"] = f"Bearer {caller_token}"
    return headers


def _context_arg(context: Context | None) -> dict[str, Context]:
    """Keep direct-call compatibility while forwarding real HTTP MCP context."""

    return {"context": context} if context is not None else {}


def _csv_env(name: str) -> list[str]:
    return [value.strip() for value in os.environ.get(name, "").split(",") if value.strip()]


def _transport_security() -> TransportSecuritySettings:
    """Keep DNS-rebinding protection on while allowing the configured public URL."""

    allowed_hosts = [
        "127.0.0.1",
        "127.0.0.1:*",
        "localhost",
        "localhost:*",
        "[::1]",
        "[::1]:*",
        "api:*",
    ]
    allowed_origins = [
        "http://127.0.0.1",
        "http://127.0.0.1:*",
        "http://localhost",
        "http://localhost:*",
        "http://[::1]",
        "http://[::1]:*",
    ]
    public_url = get_settings().public_url.strip()
    if public_url:
        parsed = urlsplit(public_url)
        if parsed.netloc:
            allowed_hosts.append(parsed.netloc)
            allowed_origins.append(f"{parsed.scheme}://{parsed.netloc}")
    allowed_hosts.extend(_csv_env("OPENCLI_MCP_ALLOWED_HOSTS"))
    allowed_origins.extend(_csv_env("OPENCLI_MCP_ALLOWED_ORIGINS"))
    return TransportSecuritySettings(
        enable_dns_rebinding_protection=True,
        allowed_hosts=list(dict.fromkeys(allowed_hosts)),
        allowed_origins=list(dict.fromkeys(allowed_origins)),
    )


mcp = MCPServer(
    "opencli-admin",
    version="0.4.1",
    instructions=(
        "For a new workflow, first inspect workflow node capabilities, then create a review-only "
        "draft from the operator's intent or preview explicit node patches, then compile it. "
        "Drafting and compilation never persist or execute work. Run only an immutable published "
        "workflow after operator review. Use source tools for collection administration."
    ),
)


async def _request(
    method: str,
    path: str,
    *,
    context: Context | None = None,
    **kwargs: Any,
) -> dict[str, Any]:
    """Call the REST API and normalize HTTP/network failures for tool callers."""

    try:
        api_base_url = _api_base_url()
        async with httpx.AsyncClient(
            base_url=api_base_url,
            timeout=30.0,
            headers=_auth_headers(context),
        ) as client:
            response = await client.request(method, path, **kwargs)
            try:
                body = response.json()
            except ValueError:
                if response.status_code >= 400:
                    return {"success": False, "error": response.text}
                response.raise_for_status()
                raise
            if response.status_code >= 400:
                error = body.get("detail") or body.get("error") or response.text
                return {"success": False, "error": error}
            return body
    except httpx.HTTPError as exc:
        return {
            "success": False,
            "error": f"request to {_api_base_url()}{path} failed: {exc}",
        }


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def list_sources(
    enabled: bool | None = None,
    channel_type: str | None = None,
    page: int = 1,
    limit: int = 20,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """List configured data sources, optionally filtered by state or channel type."""

    params: dict[str, Any] = {"page": page, "limit": limit}
    if enabled is not None:
        params["enabled"] = enabled
    if channel_type is not None:
        params["channel_type"] = channel_type
    return await _request("GET", "/api/v1/sources", **_context_arg(ctx), params=params)


@mcp.tool(annotations=WRITE_TOOL, structured_output=True)
async def create_source(
    name: str,
    channel_type: str,
    channel_config: dict[str, Any],
    description: str | None = None,
    enabled: bool = True,
    tags: list[str] | None = None,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Create an OpenCLI data source."""

    return await _request(
        "POST",
        "/api/v1/sources",
        **_context_arg(ctx),
        json={
            "name": name,
            "channel_type": channel_type,
            "channel_config": channel_config,
            "description": description,
            "enabled": enabled,
            "tags": tags or [],
        },
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def test_source(source_id: str, ctx: Context | None = None) -> dict[str, Any]:
    """Dry-run source connectivity without storing collected records."""

    return await _request(
        "POST",
        f"/api/v1/sources/{source_id}/test",
        **_context_arg(ctx),
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def discover_feed(url: str, ctx: Context | None = None) -> dict[str, Any]:
    """Find RSS/Atom feed candidates for a website."""

    return await _request(
        "POST",
        "/api/v1/sources/discover-feed",
        **_context_arg(ctx),
        json={"url": url},
    )


@mcp.tool(annotations=WRITE_TOOL, structured_output=True)
async def trigger_task(
    source_id: str,
    parameters: dict[str, Any] | None = None,
    priority: int = 5,
    agent_id: str | None = None,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Dispatch a collection run and return its task identifier."""

    return await _request(
        "POST",
        "/api/v1/tasks/trigger",
        **_context_arg(ctx),
        json={
            "source_id": source_id,
            "parameters": parameters or {},
            "priority": priority,
            "agent_id": agent_id,
        },
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def get_task(task_id: str, ctx: Context | None = None) -> dict[str, Any]:
    """Read a collection task's durable status."""

    return await _request("GET", f"/api/v1/tasks/{task_id}", **_context_arg(ctx))


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def list_records(
    source_id: str | None = None,
    task_id: str | None = None,
    status: str | None = None,
    search: str | None = None,
    page: int = 1,
    limit: int = 20,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Query collected records."""

    params: dict[str, Any] = {"page": page, "limit": limit}
    if source_id is not None:
        params["source_id"] = source_id
    if task_id is not None:
        params["task_id"] = task_id
    if status is not None:
        params["status"] = status
    if search is not None:
        params["search"] = search
    return await _request("GET", "/api/v1/records", **_context_arg(ctx), params=params)


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def query_project_records(
    workspace_id: str,
    project_id: str,
    q: str | None = None,
    limit: int = 20,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Read existing normalized records authorized for one exact project."""

    params: dict[str, Any] = {"limit": limit}
    if q is not None:
        params["q"] = q
    return await _request(
        "GET",
        f"/api/v1/workspaces/{workspace_id}/projects/{project_id}/agent-data/records",
        **_context_arg(ctx),
        params=params,
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def query_project_context(
    workspace_id: str,
    project_id: str,
    q: str,
    limit: int = 8,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Retrieve cited records, knowledge, and durable research without starting work."""

    return await _request(
        "GET",
        f"/api/v1/workspaces/{workspace_id}/projects/{project_id}/agent-data/context",
        **_context_arg(ctx),
        params={"q": q, "limit": limit},
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def get_project_data_capabilities(
    workspace_id: str,
    project_id: str,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Read truthful downstream data and research contract availability."""

    return await _request(
        "GET",
        f"/api/v1/workspaces/{workspace_id}/projects/{project_id}/agent-data/capabilities",
        **_context_arg(ctx),
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def get_project_research_readiness(
    workspace_id: str,
    project_id: str,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Inspect research dependency readiness for an authorized project."""

    return await _request(
        "GET",
        f"/api/v1/workspaces/{workspace_id}/projects/{project_id}/research/readiness",
        **_context_arg(ctx),
    )


@mcp.tool(annotations=IDEMPOTENT_WRITE_TOOL, structured_output=True)
async def start_project_research(
    workspace_id: str,
    project_id: str,
    template_id: str,
    question: str,
    request_id: str,
    seed_urls: list[str] | None = None,
    max_sources: int = 5,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Start research through the shared REST engine using a stable request ID."""

    return await _request(
        "POST",
        f"/api/v1/workspaces/{workspace_id}/projects/{project_id}/research/runs",
        **_context_arg(ctx),
        json={
            "template_id": template_id,
            "question": question,
            "seed_urls": seed_urls or [],
            "max_sources": max_sources,
            "request_id": request_id,
        },
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def list_project_research_runs(
    workspace_id: str,
    project_id: str,
    limit: int = 20,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """List durable research runs for one authorized project."""

    return await _request(
        "GET",
        f"/api/v1/workspaces/{workspace_id}/projects/{project_id}/research/runs",
        **_context_arg(ctx),
        params={"limit": limit},
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def get_project_research_run(
    workspace_id: str,
    project_id: str,
    run_id: str,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Read one durable research run after project authorization."""

    return await _request(
        "GET",
        f"/api/v1/workspaces/{workspace_id}/projects/{project_id}/research/runs/{run_id}",
        **_context_arg(ctx),
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def list_project_workflows(
    workspace_id: str,
    project_id: str,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """List a project's workflows and current published versions."""

    return await _request(
        "GET",
        f"/api/v1/workspaces/{workspace_id}/projects/{project_id}/workflows",
        **_context_arg(ctx),
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def list_workflow_node_capabilities(
    ctx: Context | None = None,
) -> dict[str, Any]:
    """List typed Workflow nodes, runtime bindings, readiness, and input/output contracts."""

    return await _request(
        "GET",
        "/api/v1/workflows/capabilities",
        **_context_arg(ctx),
    )


def _new_agent_workflow_project(intent: str, name: str, locale: str) -> dict[str, Any]:
    """Create the canonical review-only seed used by intent-driven drafting."""

    return {
        "id": "agent-workflow-draft",
        "name": name,
        "profile": "intelligence",
        "version": 1,
        "nodes": [
            {
                "id": "collection-need",
                "kind": "schedule",
                "capability": "trigger",
                "params": {"text": intent, "locale": locale, "mode": "demand-draft"},
                "proposalState": "proposed",
                "ui": {
                    "catalogId": "intelligence.input.collection-need",
                    "label": "Collection Need",
                    "position": {"x": 160, "y": 180},
                },
            }
        ],
        "edges": [],
        "adapters": [],
        "agentPermissions": {
            "canFetchNetwork": True,
            "canSendNotifications": False,
            "canWriteInbox": True,
            "canMutateExternalSites": False,
            "allowedDomains": [],
        },
    }


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def draft_workflow_from_intent(
    intent: str,
    name: str = "Agent workflow draft",
    locale: str = "zh-CN",
    project: dict[str, Any] | None = None,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Arrange existing nodes into a review-only draft; never persist, publish, or execute it."""

    base_project = project or _new_agent_workflow_project(intent, name, locale)
    return await _request(
        "POST",
        "/api/v1/workflows/demand-draft",
        **_context_arg(ctx),
        json={"project": base_project, "text": intent, "locale": locale},
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def preview_workflow_node_patch(
    project: dict[str, Any],
    operations: list[dict[str, Any]],
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Preview explicit add/connect/update node operations without persisting the graph."""

    return await _request(
        "POST",
        "/api/v1/workflows/patch",
        **_context_arg(ctx),
        json={"project": project, "operations": operations},
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def compile_workflow_draft(
    project: dict[str, Any],
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Validate and compile a draft graph in memory without dispatching or persisting work."""

    return await _request(
        "POST",
        "/api/v1/workflows/compile",
        **_context_arg(ctx),
        json={"project": project},
    )


@mcp.tool(annotations=IDEMPOTENT_WRITE_TOOL, structured_output=True)
async def run_published_workflow(
    workspace_id: str,
    project_id: str,
    workflow_id: str,
    inputs: dict[str, Any],
    idempotency_key: str,
    user: str = "mcp-client",
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Run the current immutable published version with an explicit retry key."""

    return await _request(
        "POST",
        (f"/api/v1/workspaces/{workspace_id}/projects/{project_id}/workflows/{workflow_id}/runs"),
        **_context_arg(ctx),
        headers={"Idempotency-Key": idempotency_key},
        json={"inputs": inputs, "response_mode": "async", "user": user},
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def get_project_runtime_summary(
    workspace_id: str,
    project_id: str,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Read project-level run counts and recent activity."""

    return await _request(
        "GET",
        f"/api/v1/workspaces/{workspace_id}/projects/{project_id}/runtime-summary",
        **_context_arg(ctx),
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def list_project_runtime_logs(
    workspace_id: str,
    project_id: str,
    status: str | None = None,
    search: str | None = None,
    page: int = 1,
    limit: int = 20,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """List durable project workflow runs."""

    params: dict[str, Any] = {"page": page, "limit": limit}
    if status is not None:
        params["status"] = status
    if search is not None:
        params["search"] = search
    return await _request(
        "GET",
        f"/api/v1/workspaces/{workspace_id}/projects/{project_id}/runtime-logs",
        **_context_arg(ctx),
        params=params,
    )


@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
async def get_project_runtime_trace(
    workspace_id: str,
    project_id: str,
    workflow_id: str,
    run_id: str,
    after_sequence: int | None = None,
    limit: int | None = None,
    ctx: Context | None = None,
) -> dict[str, Any]:
    """Read one project-owned run's projection, checkpoint, and events."""

    params: dict[str, Any] = {}
    if after_sequence is not None:
        params["afterSequence"] = after_sequence
    if limit is not None:
        params["limit"] = limit
    return await _request(
        "GET",
        (
            f"/api/v1/workspaces/{workspace_id}/projects/{project_id}"
            f"/workflows/{workflow_id}/runs/{run_id}/trace"
        ),
        **_context_arg(ctx),
        params=params,
    )


mcp_http_app = mcp.streamable_http_app(
    streamable_http_path="/mcp",
    json_response=True,
    stateless_http=True,
    transport_security=_transport_security(),
)


def main() -> None:
    transport = os.environ.get("OPENCLI_MCP_TRANSPORT", "stdio")
    try:
        if transport == "streamable-http":
            mcp.run(
                transport="streamable-http",
                host=os.environ.get("OPENCLI_MCP_HOST", "127.0.0.1"),
                port=int(os.environ.get("OPENCLI_MCP_PORT", "8765")),
                streamable_http_path="/mcp",
                json_response=True,
                stateless_http=True,
                transport_security=_transport_security(),
            )
            return
        if transport == "stdio":
            mcp.run(transport="stdio")
            return
        raise ValueError("OPENCLI_MCP_TRANSPORT must be 'stdio' or 'streamable-http'")
    except KeyboardInterrupt:
        raise SystemExit(130) from None


if __name__ == "__main__":
    main()