[{'Text': 'Edited point-topic-mcp/src/point_topic_mcp/tools/gbs_query_tools.py:\n\n```diff\n@@ -1,0 +1,218 @@\n+"""Raw GBS MongoDB query tool.\n+\n+Provides a single read-only tool that runs any aggregation pipeline against\n+the GBS collections (Operator, Statistics, Sources, GlobalVariables).\n+\n+This is the escape hatch: when the structured get_status/add_statistic/create_source\n+tools don\'t cover what you need, use this to explore, debug, and discover edge cases.\n+\n+Safety: only read operations allowed. Write operations ($out, $merge, etc.) are\n+explicitly rejected. Insert/update/delete handled by validated tools only.\n+"""\n+\n+import json\n+import re\n+\n+from point_topic_mcp.core.utils import check_env_vars\n+from point_topic_mcp.core import mongodb_utils as mongo\n+from dotenv import load_dotenv\n+\n+load_dotenv()\n+\n+# Collections available for read-only queries\n+ALLOWED_COLLECTIONS = {\n+    "Operator",\n+    "Statistics",\n+    "Sources",\n+    "GlobalVariables",\n+    "operator",\n+    "statistics",\n+    "sources",\n+    "globalvariables",\n+    "globalVariables",\n+}\n+\n+# Stage names that MODIFY data — reject these\n+_WRITE_STAGES = {\n+    "$out", "$merge", "$mergeObjects",\n+    "$addFields", "$set",             # $addFields/$set are actually read-only (project-like) — reconsider\n+    "$bucket", "$bucketAuto",        # These are read-only grouping stages\n+    "$changeStream", "$changeStreamSplitLargeEvent",\n+    "$collStats", "$currentOp",\n+    "$indexStats", "$listSessions",\n+    "$planCacheStats",\n+}\n+# Actually $addFields/$set ARE read-only (they add computed fields, don\'t write to DB).\n+# The real write stages are $out, $merge, $mergeObjects (top-level).\n+# $mergeObjects as an expression inside $project is fine, only top-level $merge writes.\n+# But let\'s be conservative and block the dangerous ones.\n+_WRITE_STAGES = {"$out", "$merge"}\n+\n+\n+if check_env_vars("gbs_query_tools", ["PT_RESEARCH_DATABASE_URI"]):\n+\n+    def _validate_pipeline(pipeline: list) -> None:\n+        """Validate pipeline is read-only and uses allowed stages.\n+\n+        Raises ValueError with clear message if validation fails.\n+        """\n+        if not isinstance(pipeline, list):\n+            raise ValueError("Pipeline must be a JSON array of stage objects")\n+\n+        for i, stage in enumerate(pipeline):\n+            if not isinstance(stage, dict):\n+                raise ValueError(f"Stage {i} is not a JSON object: {stage}")\n+            if not stage:\n+                raise ValueError(f"Stage {i} is empty")\n+            stage_name = next(iter(stage))\n+            if stage_name in _WRITE_STAGES:\n+                raise ValueError(\n+                    f"Stage {i} uses \'{stage_name}\' which is write-only. "\n+                    f"Only read-only aggregation pipelines are allowed."\n+                )\n+\n+    def _find_used_collections(pipeline: list) -> set:\n+        """Determine which collections the pipeline references."""\n+        used = set()\n+        for stage in pipeline:\n+            if "$lookup" in stage:\n+                from_val = stage["$lookup"].get("from", "")\n+                # from could be a dynamic expression — use the literal string\n+                if isinstance(from_val, str) and from_val:\n+                    used.add(from_val)\n+            if "$unionWith" in stage:\n+                coll = stage["$unionWith"].get("coll", "")\n+                if isinstance(coll, str) and coll:\n+                    used.add(coll)\n+            if "$facet" in stage:\n+                for facet_name, facet_pipeline in stage["$facet"].items():\n+                    used |= _find_used_collections(facet_pipeline)\n+        return used\n+\n+    def read_gbs(pipeline_json: str = "", pipeline: list = None) -> str:\n+        """Run a read-only MongoDB aggregation pipeline against GBS collections.\n+\n+        Primary collections (pipeline runs against these):\n+          - Operator      — 1300+ operators worldwide (fields: name, country, technologies, asn)\n+          - Statistics    — Subscriber counts (fields: type, operatorId, period, tech, channel,\n+                            domain, subscribers, state)\n+          - Sources       — Source URLs/files\n+          - GlobalVariables — System config (currentStatsPeriod, currentTariffsPeriod)\n+\n+        Provide either:\n+          pipeline_json: JSON string of an aggregation pipeline array\n+          pipeline: Python list (used programmatically, not by agents)\n+\n+        Examples of useful pipelines:\n+\n+        1. Get BT Group stats across all periods:\n+        ```json\n+        [\n+          {"$match": {"name": "BT Group", "country": "United Kingdom"}},\n+          {"$lookup": {\n+            "from": "Statistics",\n+            "let": {"oid": "$_id"},\n+            "pipeline": [\n+              {"$match": {"$expr": {"$eq": ["$operatorId", "$$oid"]}}}\n+            ],\n+            "as": "stats"\n+          }},\n+          {"$project": {"name": 1, "stats.period": 1, "stats.tech": 1, "stats.subscribers": 1, "stats.state": 1}}\n+        ]\n+        ```\n+\n+        2. Get current GlobalVariables:\n+        ```json\n+        [\n+          {"$match": {"isCurrent": true}},\n+          {"$project": {"currentStatsPeriod": 1, "currentTariffsPeriod": 1, "updatedBy": 1}}\n+        ]\n+        ```\n+\n+        3. Count stats by state and period:\n+        ```json\n+        [\n+          {"$group": {"_id": {"period": "$period", "state": "$state"}, "count": {"$sum": 1}}},\n+          {"$sort": {"_id.period": -1, "_id.state": 1}}\n+        ]\n+        ```\n+\n+        Returns:\n+            JSON string with pipeline results.\n+            Error messages for invalid pipelines or empty data.\n+        """\n+        # Parse pipeline\n+        if pipeline_json:\n+            try:\n+                pipeline = json.loads(pipeline_json)\n+            except json.JSONDecodeError as e:\n+                return f"Error: Invalid JSON pipeline — {e}"\n+        elif pipeline is None:\n+            return "Error: Provide pipeline_json (JSON string) or pipeline (list)"\n+\n+        # Validate read-only\n+        try:\n+            _validate_pipeline(pipeline)\n+        except ValueError as e:\n+            return f"Error: {e}"\n+\n+        # Determine which collection to run against\n+        # Default to Statistics (most useful), detect from pipeline\n+        collection = "Statistics"\n+        first_stage = pipeline[0] if pipeline else {}\n+        if "$match" in first_stage:\n+            match = first_stage["$match"]\n+            # If matching on name/country/technologies, likely going against Operator\n+            if isinstance(match, dict):\n+                match_keys = set(match.keys())\n+                if match_keys & {"name", "country", "isArchived", "technologies", "asn", "_id"}:\n+                    if "type" not in match_keys and "period" not in match_keys and "state" not in match_keys:\n+                        collection = "Operator"\n+\n+        # If $lookup references a collection, also check first stage for collection hints\n+        used_collections = _find_used_collections(pipeline)\n+        # If pipeline looks like it targets GlobalVariables\n+        if used_collections or collection == "Statistics":\n+            pass  # auto-detect below\n+\n+        # Better auto-detect: if $match keys look like a specific collection\n+        if "$match" in first_stage:\n+            match = first_stage["$match"]\n+            if isinstance(match, dict):\n+                mkeys = set(match.keys())\n+                if "isCurrent" in mkeys or "currentStatsPeriod" in mkeys:\n+                    collection = "GlobalVariables"\n+                elif "operatorId" in mkeys or "type" in mkeys or "period" in mkeys or "state" in mkeys or "subscribers" in mkeys:\n+                    collection = "Statistics"\n+                elif "year" in mkeys and "quarter" in mkeys:\n+                    collection = "Sources"\n+\n+        # Ensure target collection is allowed\n+        if collection not in ALLOWED_COLLECTIONS:\n+            return f"Error: Collection \'{collection}\' not in allowed set: {sorted(ALLOWED_COLLECTIONS)}"\n+\n+        # Run the pipeline\n+        try:\n+            result = mongo.run_aggregation(pipeline, collection=collection, timeout=60)\n+        except RuntimeError as e:\n+            return f"Query error: {e}"\n+        except json.JSONDecodeError as e:\n+            return f"Result parsing error: {e}"\n+        except Exception as e:\n+            return f"Unexpected error: {e}"\n+\n+        if not result:\n+            return "[] — no documents matched the query"\n+\n+        # Format as pretty JSON (limit output size for very large results)\n+        try:\n+            formatted = json.dumps(result, indent=2, default=str)\n+        except Exception as e:\n+            formatted = str(result)\n+\n+        # Truncate if too large (MCP message limits)\n+        MAX_CHARS = 50000\n+        if len(formatted) > MAX_CHARS:\n+            formatted = formatted[:MAX_CHARS] + f"\\n\\n... [truncated: {len(result)} total docs, showing first {formatted[:MAX_CHARS].count(chr(10))} lines]"\n+\n+        return formatted\n\n```'}]