Function _handle_mcp_request

dash/mcp/_server.py:125–168

(data)

        return _handle_mcp_request(await adapter.get_json())

    def _handle_mcp_request(data) -> Any:
        adapter = app.backend.request_adapter()
        content_type = adapter.headers.get("Content-Type", "")
        if "application/json" not in content_type:
            return app.backend.make_response(
                json.dumps({"error": "Content-Type must be application/json"}),
                content_type="application/json",
                status=415,
            )

        if data is None:
            return app.backend.make_response(
                json.dumps({"error": "Invalid JSON"}),
                content_type="application/json",
                status=400,
            )

        method = data.get("method", "")

        try:
            is_stale_session = _check_session(method)
        except ValueError as err:
            return app.backend.make_response(
                json.dumps({"error": str(err)}),
                content_type="application/json",
                status=400,
            )

        response_data = _process_mcp_message(data)

        if response_data is None:
            return app.backend.make_response("", status=202)

        if is_stale_session:
            return _json_response(
                {"jsonrpc": "2.0", "method": "notifications/tools/list_changed"},
                {
                    "jsonrpc": "2.0",
                    "method": "notifications/resources/list_changed",
                },
                response_data,
            )

        return _json_response(response_data)

    def _handle_get():
        """