Nexum-Expanded/runtime/src/nexum_runtime/protocols.py

"""Portable protocol projections over the Nexum runtime boundary."""

from __future__ import annotations

import json
import uuid
from typing import Any, Callable

from nexum_core.api import NEXUM_CORE_VERSION

from .capabilities import capabilities
from .executor import advertised_tool_spec, execute_tool_call, runtime_tool_schemas
from .tooling.contracts import tool_call_from_openai


def agent_card(base_url: str) -> dict[str, Any]:
    skills = [
        {
            "id": row.capability_id,
            "name": row.name,
            "description": row.description,
            "tags": list(row.surfaces),
        }
        for row in capabilities()
        if row.available
    ]
    return {
        "name": "Nexum",
        "description": "Private universal tool model served through NNF X.",
        "url": base_url.rstrip("/") + "/a2a",
        "version": NEXUM_CORE_VERSION,
        "protocolVersion": "0.3.0",
        "capabilities": {
            "streaming": False,
            "pushNotifications": False,
            "stateTransitionHistory": True,
        },
        "defaultInputModes": ["text/plain", "application/json"],
        "defaultOutputModes": ["text/plain", "application/json"],
        "skills": skills,
    }


def mcp_tools() -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for schema in runtime_tool_schemas():
        function = schema.get("function")
        if not isinstance(function, dict):
            continue
        rows.append(
            {
                "name": str(function.get("name") or ""),
                "description": str(function.get("description") or ""),
                "inputSchema": function.get("parameters") or {"type": "object"},
                "annotations": schema.get("x-nexum") or {},
            }
        )
    return rows


def mcp_request(
    payload: dict[str, Any],
    *,
    workspace: str,
    session_id: str,
    tools_enabled: bool,
) -> dict[str, Any]:
    request_id = payload.get("id")
    method = str(payload.get("method") or "")

    def success(result: dict[str, Any]) -> dict[str, Any]:
        return {"jsonrpc": "2.0", "id": request_id, "result": result}

    if method == "initialize":
        return success(
            {
                "protocolVersion": "2025-11-25",
                "capabilities": {"tools": {"listChanged": False}},
                "serverInfo": {
                    "name": "Nexum NNF X",
                    "version": NEXUM_CORE_VERSION,
                },
            }
        )
    if method == "ping":
        return success({})
    if method == "tools/list":
        return success({"tools": mcp_tools()})
    if method == "tools/call":
        if not tools_enabled:
            raise PermissionError("server tool execution is disabled")
        params = payload.get("params")
        if not isinstance(params, dict):
            raise ValueError("tools/call params must be an object")
        call_id = str(params.get("call_id") or "call_" + uuid.uuid4().hex)
        call = tool_call_from_openai(
            {
                "id": call_id,
                "type": "function",
                "function": {
                    "name": str(params.get("name") or ""),
                    "arguments": params.get("arguments") or {},
                },
                "x-nexum": params.get("x-nexum") or {},
            }
        )
        selected_spec = advertised_tool_spec(call.name)
        if selected_spec is not None and selected_spec.execution_owner != "runtime":
            raise ValueError("caller-owned tools cannot execute through local MCP")
        result = execute_tool_call(
            call,
            cwd=workspace,
            session_id=session_id,
        )
        payload_result = result.to_dict()
        return success(
            {
                "content": [{"type": "text", "text": json.dumps(payload_result, sort_keys=True)}],
                "structuredContent": payload_result,
                "isError": not result.ok,
            }
        )
    raise ValueError(f"unsupported MCP method: {method}")


def a2a_request(
    payload: dict[str, Any],
    *,
    run_agent: Callable[[dict[str, Any]], dict[str, Any]],
) -> dict[str, Any]:
    request_id = payload.get("id")
    method = str(payload.get("method") or "")
    if method != "message/send":
        raise ValueError(f"unsupported delegated-agent method: {method}")
    params = payload.get("params")
    if not isinstance(params, dict):
        raise ValueError("delegated-agent params must be an object")
    message = params.get("message")
    if not isinstance(message, dict):
        raise ValueError("message is required")
    parts = message.get("parts")
    text = ""
    if isinstance(parts, list):
        text = "\n".join(
            str(part.get("text") or "")
            for part in parts
            if isinstance(part, dict) and part.get("kind") in {None, "text"}
        )
    response = run_agent(
        {
            "prompt": text,
            "session_id": str(params.get("contextId") or uuid.uuid4().hex),
        }
    )
    pending_calls = response.get("tool_calls")
    if not isinstance(pending_calls, (list, tuple)):
        pending_calls = ()
    if response.get("message") == "completed":
        state = "completed"
    elif pending_calls:
        state = "input-required"
    else:
        state = "working"
    metadata = dict(response.get("metadata") or {})
    if pending_calls:
        metadata["pending_tool_calls"] = [
            dict(call) for call in pending_calls if isinstance(call, dict)
        ]
    result = {
        "id": "task_" + uuid.uuid4().hex,
        "contextId": str(params.get("contextId") or ""),
        "status": {"state": state},
        "artifacts": (
            [
                {
                    "artifactId": "artifact_" + uuid.uuid4().hex,
                    "parts": [{"kind": "text", "text": str(response.get("final_text") or "")}],
                }
            ]
            if state == "completed"
            else []
        ),
        "metadata": metadata,
    }
    return {"jsonrpc": "2.0", "id": request_id, "result": result}


__all__ = ["a2a_request", "agent_card", "mcp_request", "mcp_tools"]