tools/mcp_server.py

"""MCP server — the in-process FastMCP backend's actual tools.

WHAT THIS IS
------------
This is a small **FastMCP server** that exposes a handful of SOC tools over the
Model Context Protocol (MCP). MCP is the open standard a model/agent uses to call
out to an external "tool server". Here that server is the thing an agent would
talk to in production (a SIEM, a CMDB, a ticketing system), but in this demo it
runs **in-process over an in-memory transport** — there is no network socket and
no external endpoint, which keeps the demo safe and fully self-contained.

THE TOOLS IT EXPOSES (names match the policy exactly — policies/agt_policy.yaml)
--------------------------------------------------------------------------------
    * search_alerts        — find SIEM alerts (read-only)
    * get_asset_context    — fetch a host + its containers (read-only)
    * create_ticket_draft  — draft a ticket as text; persists NOTHING (advisory)
    * create_ticket        — actually create a ticket in the mock store (a write)

WHERE IT SITS IN THE TRUST MODEL (same as every backend)
--------------------------------------------------------
This server never decides *whether* an action is allowed. That already happened
upstream: by the time the MCP client backend calls one of these tools, the gate
has said ``ALLOW`` and the dispatcher has verified the gate's signed grant.

    gate decides ─► dispatcher verifies grant ─► mcp_client backend ─► THIS server runs

WHY A FACTORY (``build_server``) INSTEAD OF A MODULE-LEVEL SERVER
----------------------------------------------------------------
Each call to :func:`build_server` returns a *fresh* server with its own private
ticket store (a deep copy of the fixtures). That means tests — and separate demo
runs — never leak created tickets into one another, exactly like the Direct-API
backend's per-instance store.
"""

from __future__ import annotations

from typing import Any

from fastmcp import FastMCP

import data

# The four tool names this server implements. Re-exported so the client backend
# can use it as its defense-in-depth allow-list without re-typing the names.
MCP_TOOL_NAMES: tuple[str, ...] = (
    "search_alerts",
    "get_asset_context",
    "create_ticket_draft",
    "create_ticket",
)


class _McpToolState:
    """The data the server's tools read and write — one private copy per server.

    Reads (alerts, assets) come from the shared cached fixtures; the ticket store
    is a deep copy so ``create_ticket`` can add records without ever mutating the
    on-disk seed other components rely on.
    """

    def __init__(self) -> None:
        self.alerts = data.load_alerts()          # read-only list of SIEM alerts
        self.assets = data.index_by_id("assets")  # read-only, keyed by asset_id
        # The one writable store, keyed by ticket_id (deep copy → isolated).
        self.tickets: dict[str, dict[str, Any]] = {
            t["ticket_id"]: t for t in data.load_records("tickets")
        }

    def next_ticket_id(self) -> str:
        """Mint the next ``TKT-####`` id, one past the highest currently stored."""
        numbers = [
            int(tid.split("-")[1])
            for tid in self.tickets
            if tid.startswith("TKT-") and tid.split("-")[1].isdigit()
        ]
        return f"TKT-{(max(numbers, default=5000) + 1):04d}"


def build_server() -> FastMCP:
    """Build a fresh in-process FastMCP server with its own ticket store.

    The tools are registered as closures over a per-server :class:`_McpToolState`,
    so two servers built here never share writable state. The returned server is
    not "started" in any network sense — the client connects to it over an
    in-memory transport (see ``execution_backends/mcp_client.py``).
    """
    state = _McpToolState()
    server = FastMCP(name="soc-mcp-tools")

    # NOTE: every tool is decorated with @server.tool, which inspects the type
    # hints below to publish a JSON schema to MCP clients. Returning plain dicts
    # gives the client structured content it can read back as ``result.data``.

    @server.tool
    def search_alerts(
        incident_id: str | None = None,
        severity: str | None = None,
        status: str | None = None,
    ) -> dict[str, Any]:
        """Search SIEM alerts, optionally filtered by incident, severity, or status.

        Read-only: it only filters the mock alert list and never changes anything.
        All filters are optional and combine with AND; omitting them returns every
        alert.
        """
        matches = [
            alert
            for alert in state.alerts
            if (incident_id is None or alert.get("incident_id") == incident_id)
            and (severity is None or alert.get("severity") == severity)
            and (status is None or alert.get("status") == status)
        ]
        return {"count": len(matches), "alerts": matches}

    @server.tool
    def get_asset_context(asset_id: str) -> dict[str, Any]:
        """Return the full context for one asset: the host record and its containers.

        Read-only. Raises if the asset id is unknown rather than returning an empty
        result, so a typo surfaces as a clear error instead of silent "no data".
        """
        asset = state.assets.get(asset_id)
        if asset is None:
            raise ValueError(f"no asset with id {asset_id!r}")
        return asset

    @server.tool
    def create_ticket_draft(
        title: str,
        incident_id: str | None = None,
        asset_id: str | None = None,
        body: str | None = None,
    ) -> dict[str, Any]:
        """Draft a ticket as text — **persists nothing**.

        This is the L1 "advisory" MCP action: it assembles what a ticket *would*
        look like for a human to review, but it does not add anything to the store.
        The ``is_draft`` flag and absence of a ``ticket_id`` make that explicit.
        """
        return {
            "kind": "ticket_draft",
            "is_draft": True,
            "incident_id": incident_id,
            "asset_id": asset_id,
            "title": title,
            "body": body or f"Draft ticket for review: {title}",
        }

    @server.tool
    def create_ticket(
        title: str,
        incident_id: str | None = None,
        asset_id: str | None = None,
        priority: str = "low",
        assignee_user_id: str | None = None,
    ) -> dict[str, Any]:
        """Create a ticket in the mock store and return it (a bounded, reversible write).

        This is the L2 action: it actually adds a record. Like the Direct-API
        backend, it only ever creates *low-risk* tickets, which is what keeps it a
        reversible, policy-allowed bounded action.
        """
        ticket = {
            "ticket_id": state.next_ticket_id(),
            "incident_id": incident_id,
            "title": title,
            "status": "open",
            "priority": priority,
            "risk": "low",  # this tool only ever creates low-risk tickets
            "asset_id": asset_id,
            "assignee_user_id": assignee_user_id,
            "comments": [],
        }
        state.tickets[ticket["ticket_id"]] = ticket
        return ticket

    return server