Function _process_mcp_message

dash/mcp/_server.py:250–311

Process an MCP JSON-RPC message and return the response dict. Returns ``None`` for notifications (no ``id`` field).

(data: dict[str, Any])

def _process_mcp_message(data: dict[str, Any]) -> dict[str, Any] | None:
    """
    Process an MCP JSON-RPC message and return the response dict.

    Returns ``None`` for notifications (no ``id`` field).
    """
    method = data.get("method", "")
    params = data.get("params", {}) or {}
    _id = data.get("id")
    request_id: str | int = _id if isinstance(_id, (str, int)) else ""

    app = get_app()
    if app.mcp_callback_map is None:
        app.mcp_callback_map = CallbackAdapterCollection(app)
        app.mcp_decorated_functions = dict(MCP_DECORATED_FUNCTIONS)

    mcp_methods = {
        "initialize": _handle_initialize,
        "tools/list": list_tools,
        "tools/call": lambda: call_tool(
            tool_name=params.get("name", ""),
            arguments=params.get("arguments", {}),
            task=params.get("task"),
        ),
        "resources/list": list_resources,
        "resources/templates/list": list_resource_templates,
        "resources/read": lambda: read_resource(params.get("uri", "")),
        "tasks/get": lambda: get_task(task_id=params.get("taskId", "")),
        "tasks/result": lambda: get_task_result(task_id=params.get("taskId", "")),
        "tasks/cancel": lambda: cancel_task(task_id=params.get("taskId", "")),
    }

    try:
        handler = mcp_methods.get(method)
        if handler is None:
            if method.startswith("notifications/"):
                return None
            raise ValueError(f"Unknown method: {method}")

        result = handler()

        response = JSONRPCResponse(
            jsonrpc="2.0",
            id=request_id,
            result=result.model_dump(exclude_none=True, mode="json"),
        )
        return response.model_dump(exclude_none=True, mode="json")

    except MCPError as e:
        logger.error("MCP error: %s", e)
        return JSONRPCError(
            jsonrpc="2.0",
            id=request_id,
            error=ErrorData(code=e.code, message=str(e)),
        ).model_dump(exclude_none=True)
    except Exception as e:  # pylint: disable=broad-exception-caught
        logger.error("MCP error: %s", e, exc_info=True)
        return JSONRPCError(