Build your own MCP server.

"""Reference MCP adapter for Commonset's governed REST API.

Read methods inspect authorized versions; write methods create private drafts,
create new immutable versions or submit a version for human review. Every write
requires explicit user intent and the appropriate Commonset permissions.
No capability is executed, published or approved by this adapter.
"""
from __future__ import annotations

import os
from urllib.parse import quote, urlparse

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("commonset-custom")
API_BASE = os.getenv("COMMONSET_API_BASE_URL", "https://app.commonset.ai/api/v1/").rstrip("/") + "/"


def _api_base() -> str:
    parsed = urlparse(API_BASE)
    if parsed.scheme != "https" and not (
        parsed.scheme == "http" and parsed.hostname in {"localhost", "127.0.0.1"}
    ):
        raise ValueError("COMMONSET_API_BASE_URL must use HTTPS (except localhost).")
    if not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment:
        raise ValueError("Invalid COMMONSET_API_BASE_URL.")
    return API_BASE


def _credential() -> str:
    token = os.getenv("COMMONSET_API_TOKEN", "").strip()
    if not token or not token.startswith("csmcp_"):
        raise RuntimeError("Set COMMONSET_API_TOKEN to a Commonset developer token.")
    return token


async def _request(method: str, path: str, *, params: dict | None = None, payload: dict | None = None) -> dict:
    """Never redirect bearer tokens or echo headers/secret material in errors."""
    base = _api_base()
    token = _credential()
    async with httpx.AsyncClient(
        base_url=base,
        timeout=20.0,
        follow_redirects=False,
        headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
    ) as client:
        try:
            if method == "GET":
                response = await client.get(path, params=params or {})
            elif method == "POST":
                response = await client.post(path, json=payload or {})
            else:
                raise ValueError("Unsupported Commonset method.")
            response.raise_for_status()
        except httpx.HTTPStatusError as exc:
            status = exc.response.status_code
            if status == 409:
                explanation = "Version or review conflict. Fetch current managed/status metadata before retrying."
            elif status == 404:
                explanation = "The endpoint or requested capability/version is unavailable."
            elif status == 429:
                explanation = "Read or write rate limit reached. Retry only after checking the current state."
            else:
                explanation = "Check developer token, permissions and request payload."
            raise RuntimeError(f"Commonset returned HTTP {status}. {explanation}") from None
        except httpx.RequestError:
            raise RuntimeError("Could not reach the Commonset API.") from None
        return response.json()


def _slug(value: str) -> str:
    if not value or "/" in value or "\\" in value or ".." in value:
        raise ValueError("Supply a single Commonset capability slug.")
    return quote(value, safe="")


@mcp.tool()
async def list_commonset_capabilities(query: str = "", limit: int = 25) -> dict:
    """List trusted, user-authorized versions. Creator drafts are not necessarily approved."""
    if not 1 <= limit <= 50:
        raise ValueError("limit must be between 1 and 50.")
    return await _request("GET", "capabilities/", params={"limit": limit, **({"q": query} if query else {})})


@mcp.tool()
async def get_commonset_capability(slug: str, version: str = "") -> dict:
    """Read the exact trusted version's metadata and native entrypoint instructions."""
    return await _request("GET", f"capabilities/{_slug(slug)}/", params={"version": version} if version else {})


@mcp.tool()
async def get_commonset_package(slug: str, version: str = "") -> dict:
    """Read verified files with hashes. Never execute downloaded scripts automatically."""
    return await _request("GET", f"capabilities/{_slug(slug)}/package/", params={"version": version} if version else {})


@mcp.tool()
async def list_managed_commonset_capabilities(limit: int = 25) -> dict:
    """Find editable capabilities and the latest immutable version UUID for stale-write protection."""
    if not 1 <= limit <= 50:
        raise ValueError("limit must be between 1 and 50.")
    return await _request("GET", "managed-capabilities/", params={"limit": limit})


@mcp.tool()
async def get_commonset_capability_status(slug: str, version: str = "") -> dict:
    """Inspect review, trust, creator availability and publication gate without changing state."""
    return await _request(
        "GET", f"capabilities/{_slug(slug)}/status/",
        params={"version": version} if version else {},
    )


@mcp.tool()
async def create_commonset_draft(
    name: str, description: str, instructions: str,
    slug: str = "", capability_type: str = "skill",
    files: list[dict[str, str]] | None = None,
) -> dict:
    """WRITE: only when explicitly requested, create a private Commonset capability draft.

    Security analysis, organization binding and creator ownership are enforced
    by Commonset. This does not submit, approve or publish the capability.
    """
    payload = {
        "name": name, "description": description, "instructions": instructions,
        "capability_type": capability_type, "visibility": "private",
    }
    if slug:
        payload["slug"] = slug
    if files:
        payload["files"] = files
    return await _request("POST", "capabilities/", payload=payload)


@mcp.tool()
async def create_commonset_version_draft(
    slug: str, base_version_id: str, instructions: str = "",
    description: str = "", release_notes: str = "",
    files: list[dict[str, str]] | None = None,
) -> dict:
    """WRITE: only when explicitly requested, propose a new immutable capability version.

    Obtain base_version_id from managed capabilities first. Commonset refuses
    a stale base rather than overwriting approved or newly created versions.
    """
    payload = {"base_version": base_version_id}
    if instructions:
        payload["instructions"] = instructions
    if description:
        payload["description"] = description
    if release_notes:
        payload["release_notes"] = release_notes
    if files:
        payload["files"] = files
    return await _request("POST", f"capabilities/{_slug(slug)}/versions/", payload=payload)


@mcp.tool()
async def submit_commonset_version_for_review(slug: str, version_id: str) -> dict:
    """WRITE: only when explicitly requested, submit an exact version to human review.

    Does not approve, publish or execute. Duplicate open review submissions
    fail with a conflict rather than opening a duplicate round.
    """
    return await _request(
        "POST", f"capabilities/{_slug(slug)}/versions/{quote(version_id, safe='')}/reviews/",
        payload={},
    )


if __name__ == "__main__":
    mcp.run(transport="stdio")

httpx>=0.27,<1
mcp>=1.0,<2