feat: add read-only discovery and admin tools (9 tools)

analytics_mcp/coordinator.py

@@ -29,10 +29,18 @@
 from google.adk.tools.function_tool import FunctionTool
 from google.adk.tools.mcp_tool.conversion_utils import adk_to_mcp_tool_type
 
+from analytics_mcp.tools.admin.access import run_access_report
 from analytics_mcp.tools.admin.info import (
     get_account_summaries,
+    get_data_retention_settings,
     list_google_ads_links,
     get_property_details,
+    list_audiences,
+    list_custom_dimensions,
+    list_custom_metrics,
+    list_data_streams,
+    list_key_events,
+    list_properties,
     list_property_annotations,
 )
 from analytics_mcp.tools.reporting.core import (
@@ -45,7 +53,9 @@
 )
 from analytics_mcp.tools.reporting.metadata import (
     get_custom_dimensions_and_metrics,
+    get_metadata,
 )
+from analytics_mcp.tools.reporting.quotas import get_property_quotas
 from analytics_mcp.tools.reporting.funnel import (
     run_funnel_report,
     _run_funnel_report_description,
@@ -75,8 +85,18 @@
     FunctionTool(get_account_summaries),
     FunctionTool(list_google_ads_links),
     FunctionTool(get_property_details),
+    FunctionTool(get_data_retention_settings),
+    FunctionTool(list_audiences),
+    FunctionTool(list_custom_dimensions),
+    FunctionTool(list_custom_metrics),
+    FunctionTool(list_data_streams),
+    FunctionTool(list_key_events),
+    FunctionTool(list_properties),
     FunctionTool(list_property_annotations),
+    FunctionTool(run_access_report),
     FunctionTool(get_custom_dimensions_and_metrics),
+    FunctionTool(get_metadata),
+    FunctionTool(get_property_quotas),
     run_report_with_description,
     run_realtime_report_with_description,
     run_funnel_report_with_description,
@@ -142,6 +162,13 @@ def sanitize_mcp_schema_properties(node: dict) -> None:
         ]
     elif tool.name == "run_realtime_report":
         tool.inputSchema["required"] = ["property_id", "dimensions", "metrics"]
+    elif tool.name == "run_access_report":
+        tool.inputSchema["required"] = [
+            "entity",
+            "date_ranges",
+            "dimensions",
+            "metrics",
+        ]
     elif tool.name == "run_conversions_report":
         tool.inputSchema["required"] = [
             "property_id",

analytics_mcp/tools/admin/access.py

@@ -0,0 +1,99 @@
+# Copyright 2025 Google LLC All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tools for running data access reports using the Admin API."""
+
+import asyncio
+from typing import Any, Dict, List
+
+from analytics_mcp.tools.utils import (
+    construct_account_rn,
+    construct_property_rn,
+    proto_to_dict,
+)
+from analytics_mcp.tools.client import create_admin_api_client
+from google.analytics import admin_v1beta
+
+
+def _construct_access_report_entity(entity: int | str) -> str:
+    """Returns the entity resource name for an access report request.
+
+    Accepts a property (number or 'properties/N') or an account
+    ('accounts/N').
+    """
+    if isinstance(entity, str) and entity.strip().startswith("accounts/"):
+        return construct_account_rn(entity)
+    return construct_property_rn(entity)
+
+
+async def run_access_report(
+    entity: int | str,
+    date_ranges: List[Dict[str, str]],
+    dimensions: List[str],
+    metrics: List[str],
+    limit: int = None,
+    offset: int = None,
+    return_entity_quota: bool = False,
+) -> Dict[str, Any]:
+    """Runs a data access report showing who accessed Analytics data.
+
+    Each row describes data access events, e.g. which user accessed
+    which property, when, from where, and how many data requests were
+    made. Useful for compliance and security auditing.
+
+    Access reports use their own dimension and metric names, documented
+    at
+    https://developers.google.com/analytics/devguides/config/admin/v1/access-api-schema.
+    Common dimensions include `userEmail`, `epochTimeMicros`,
+    `accessMechanism`, and `reportType`. Common metrics include
+    `accessCount`.
+
+    Args:
+        entity: The entity to report on. Accepted formats are:
+          - A number (treated as a property ID)
+          - A string consisting of 'properties/' followed by a number
+          - A string consisting of 'accounts/' followed by a number,
+            which reports on all properties under the account
+        date_ranges: A list of date range dicts, each with `start_date`
+          and `end_date` keys. Dates are in `YYYY-MM-DD` format, or the
+          relative forms `today`, `yesterday`, and `NdaysAgo`.
+        dimensions: A list of access report dimension names.
+        metrics: A list of access report metric names.
+        limit: The maximum number of rows to return (max 100,000).
+        offset: The row count of the start row (0-indexed).
+        return_entity_quota: Whether to include the current state of
+          this property's access report quota in the response.
+    """
+    request = admin_v1beta.RunAccessReportRequest(
+        entity=_construct_access_report_entity(entity),
+        date_ranges=[admin_v1beta.AccessDateRange(dr) for dr in date_ranges],
+        dimensions=[
+            admin_v1beta.AccessDimension(dimension_name=d) for d in dimensions
+        ],
+        metrics=[admin_v1beta.AccessMetric(metric_name=m) for m in metrics],
+        return_entity_quota=return_entity_quota,
+    )
+
+    if limit:
+        request.limit = limit
+
+    if offset:
+        request.offset = offset
+
+    def _sync_call():
+        return create_admin_api_client().run_access_report(request=request)
+
+    response = await asyncio.to_thread(_sync_call)
+
+    return proto_to_dict(response)

analytics_mcp/tools/admin/info.py

@@ -18,6 +18,7 @@
 from typing import Any, Dict, List
 
 from analytics_mcp.tools.utils import (
+    construct_account_rn,
     construct_property_rn,
     proto_to_dict,
 )
@@ -78,6 +79,202 @@ def _sync_call():
     return proto_to_dict(response)
 
 
+async def list_properties(
+    account_id: int | str, show_deleted: bool = False
+) -> List[Dict[str, Any]]:
+    """Returns the properties under a Google Analytics account.
+
+    Returns full property objects, including display name, industry
+    category, time zone, currency code, service level, and create time.
+    Use this for deeper property discovery than `get_account_summaries`,
+    which only returns summary information.
+
+    Args:
+        account_id: The Google Analytics account ID. Accepted formats are:
+          - A number
+          - A string consisting of 'accounts/' followed by a number
+        show_deleted: Whether to include soft-deleted (i.e. "trashed")
+          properties in the results. Defaults to False.
+    """
+    request = admin_v1beta.ListPropertiesRequest(
+        filter=f"parent:{construct_account_rn(account_id)}",
+        show_deleted=show_deleted,
+    )
+
+    def _sync_call():
+        properties_pager = create_admin_api_client().list_properties(
+            request=request
+        )
+        return [proto_to_dict(prop) for prop in properties_pager]
+
+    return await asyncio.to_thread(_sync_call)
+
+
+async def list_key_events(property_id: int | str) -> List[Dict[str, Any]]:
+    """Returns the key events configured for a property.
+
+    Key events (formerly known as conversion events) are the events that a
+    property has marked as most important, such as purchases or sign-ups.
+    Reports use key events to calculate conversion-related metrics.
+
+    Args:
+        property_id: The Google Analytics property ID. Accepted formats are:
+          - A number
+          - A string consisting of 'properties/' followed by a number
+    """
+    request = admin_v1beta.ListKeyEventsRequest(
+        parent=construct_property_rn(property_id)
+    )
+
+    def _sync_call():
+        key_events_pager = create_admin_api_client().list_key_events(
+            request=request
+        )
+        return [proto_to_dict(key_event) for key_event in key_events_pager]
+
+    return await asyncio.to_thread(_sync_call)
+
+
+async def list_data_streams(property_id: int | str) -> List[Dict[str, Any]]:
+    """Returns the data streams configured for a property.
+
+    Data streams are the sources that send data to a property, such as a
+    web site, an Android app, or an iOS app. Each stream includes its
+    platform-specific details, e.g. the measurement ID for a web stream.
+
+    Args:
+        property_id: The Google Analytics property ID. Accepted formats are:
+          - A number
+          - A string consisting of 'properties/' followed by a number
+    """
+    request = admin_v1beta.ListDataStreamsRequest(
+        parent=construct_property_rn(property_id)
+    )
+
+    def _sync_call():
+        data_streams_pager = create_admin_api_client().list_data_streams(
+            request=request
+        )
+        return [
+            proto_to_dict(data_stream) for data_stream in data_streams_pager
+        ]
+
+    return await asyncio.to_thread(_sync_call)
+
+
+async def list_audiences(property_id: int | str) -> List[Dict[str, Any]]:
+    """Returns the audiences defined on a property.
+
+    Audiences are segments of users, e.g. "Purchasers" or "Users who
+    visited the pricing page". Knowing what audiences exist helps when
+    interpreting audience-scoped dimensions in reports or suggesting
+    targeting strategies.
+
+    Note: this uses the alpha channel of the Admin API, which may
+    change without notice.
+
+    Args:
+        property_id: The Google Analytics property ID. Accepted formats are:
+          - A number
+          - A string consisting of 'properties/' followed by a number
+    """
+    request = admin_v1alpha.ListAudiencesRequest(
+        parent=construct_property_rn(property_id)
+    )
+
+    def _sync_call():
+        audiences_pager = create_admin_alpha_api_client().list_audiences(
+            request=request
+        )
+        return [proto_to_dict(audience) for audience in audiences_pager]
+
+    return await asyncio.to_thread(_sync_call)
+
+
+async def get_data_retention_settings(
+    property_id: int | str,
+) -> Dict[str, Any]:
+    """Returns the data retention settings for a property.
+
+    Shows how long event-level and user-level data is kept before being
+    automatically deleted. Useful for compliance questions and for
+    knowing how far back reports can reliably look: if retention is set
+    to TWO_MONTHS, event-scoped data older than that is unavailable in
+    explorations and some reports.
+
+    Args:
+        property_id: The Google Analytics property ID. Accepted formats are:
+          - A number
+          - A string consisting of 'properties/' followed by a number
+    """
+    request = admin_v1beta.GetDataRetentionSettingsRequest(
+        name=f"{construct_property_rn(property_id)}/dataRetentionSettings"
+    )
+
+    def _sync_call():
+        client = create_admin_api_client()
+        return client.get_data_retention_settings(request=request)
+
+    response = await asyncio.to_thread(_sync_call)
+    return proto_to_dict(response)
+
+
+async def list_custom_dimensions(
+    property_id: int | str,
+) -> List[Dict[str, Any]]:
+    """Returns the custom dimensions defined on a property (Admin API).
+
+    Includes details that the Data API metadata endpoint doesn't return:
+    the dimension's scope (EVENT or USER), parameter name, description,
+    and whether it's excluded from ads personalization. Use the
+    `get_custom_dimensions_and_metrics` tool instead if you only need
+    the API names for building reports.
+
+    Args:
+        property_id: The Google Analytics property ID. Accepted formats are:
+          - A number
+          - A string consisting of 'properties/' followed by a number
+    """
+    request = admin_v1beta.ListCustomDimensionsRequest(
+        parent=construct_property_rn(property_id)
+    )
+
+    def _sync_call():
+        dimensions_pager = create_admin_api_client().list_custom_dimensions(
+            request=request
+        )
+        return [proto_to_dict(dimension) for dimension in dimensions_pager]
+
+    return await asyncio.to_thread(_sync_call)
+
+
+async def list_custom_metrics(property_id: int | str) -> List[Dict[str, Any]]:
+    """Returns the custom metrics defined on a property (Admin API).
+
+    Includes details that the Data API metadata endpoint doesn't return:
+    the metric's scope, parameter name, description, measurement unit,
+    and restricted metric type. Use the
+    `get_custom_dimensions_and_metrics` tool instead if you only need
+    the API names for building reports.
+
+    Args:
+        property_id: The Google Analytics property ID. Accepted formats are:
+          - A number
+          - A string consisting of 'properties/' followed by a number
+    """
+    request = admin_v1beta.ListCustomMetricsRequest(
+        parent=construct_property_rn(property_id)
+    )
+
+    def _sync_call():
+        metrics_pager = create_admin_api_client().list_custom_metrics(
+            request=request
+        )
+        return [proto_to_dict(metric) for metric in metrics_pager]
+
+    return await asyncio.to_thread(_sync_call)
+
+
 async def list_property_annotations(
     property_id: int | str,
 ) -> List[Dict[str, Any]]: