feat: added funnel report generation

Author: Arunabh Banerjee

analytics_mcp/tools/reporting/core.py

@@ -14,7 +14,7 @@
 
 """Tools for running core reports using the Data API."""
 
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 from analytics_mcp.coordinator import mcp
 from analytics_mcp.tools.reporting.metadata import (
@@ -26,9 +26,10 @@
 from analytics_mcp.tools.utils import (
     construct_property_rn,
     create_data_api_client,
+    create_data_api_alpha_client,
     proto_to_dict,
 )
-from google.analytics import data_v1beta
+from google.analytics import data_v1beta, data_v1alpha
 
 
 def _run_report_description() -> str:
@@ -79,6 +80,112 @@ def _run_report_description() -> str:
           """
 
 
+def _run_funnel_report_description() -> str:
+    """Returns the description for the `run_funnel_report` tool."""
+    return f"""
+          {run_funnel_report.__doc__}
+
+          ## Hints for arguments
+
+          Here are some hints that outline the expected format and requirements
+          for arguments.
+
+          ### Hints for `funnel_steps`
+
+          The `funnel_steps` list must contain at least 2 steps. Each step can be configured in two ways:
+
+          1. **Simple Event-Based Steps**: For basic event filtering
+             ```json
+             {{
+                 "name": "Step Name",
+                 "event": "event_name"
+             }}
+             ```
+
+          2. **Advanced Filter Expression Steps**: For complex filtering with multiple conditions
+             ```json
+             {{
+                 "name": "Step Name",
+                 "filter_expression": {{
+                     "funnel_field_filter": {{
+                         "field_name": "eventName",
+                         "string_filter": {{
+                             "match_type": "EXACT",
+                             "value": "page_view"
+                         }}
+                     }}
+                 }}
+             }}
+             ```
+
+          For page path filtering, use:
+          ```json
+          {{
+              "name": "Home Page View",
+              "filter_expression": {{
+                  "and_group": {{
+                      "expressions": [
+                          {{
+                              "funnel_field_filter": {{
+                                  "field_name": "eventName",
+                                  "string_filter": {{"match_type": "EXACT", "value": "page_view"}}
+                              }}
+                          }},
+                          {{
+                              "funnel_field_filter": {{
+                                  "field_name": "pagePath",
+                                  "string_filter": {{"match_type": "EXACT", "value": "/"}}
+                              }}
+                          }}
+                      ]
+                  }}
+              }}
+          }}
+          ```
+
+          ### Hints for `date_ranges`:
+          {get_date_ranges_hints()}
+
+          ### Hints for `funnel_breakdown`
+
+          The `funnel_breakdown` parameter allows you to segment funnel results by a dimension:
+          ```json
+          {{
+              "breakdown_dimension": "deviceCategory"
+          }}
+          ```
+
+          Common breakdown dimensions include:
+          - `deviceCategory` - Desktop, Mobile, Tablet
+          - `country` - User's country
+          - `operatingSystem` - User's operating system
+          - `browser` - User's browser
+
+          ### Hints for `funnel_next_action`
+
+          The `funnel_next_action` parameter analyzes what users do after completing or dropping off from the funnel:
+          ```json
+          {{
+              "next_action_dimension": "eventName",
+              "limit": 5
+          }}
+          ```
+
+          Common next action dimensions include:
+          - `eventName` - Next events users trigger
+          - `pagePath` - Next pages users visit
+
+          ### Important Notes
+
+          - The runFunnelReport method is currently in **alpha** status and may change
+          - Funnel reports require at least 2 steps to be meaningful
+          - Use `return_property_quota: true` to monitor your API usage
+          - For complex filtering, prefer `filter_expression` over simple `event` filters
+          - Date ranges support relative dates like "7daysAgo", "today", "yesterday"
+
+          """
+
+
 async def run_report(
     property_id: int | str,
     date_ranges: List[Dict[str, str]],
@@ -172,6 +279,218 @@ async def run_report(
 
     return proto_to_dict(response)
 
+async def run_funnel_report(
+    property_id: int | str,
+    funnel_steps: List[Dict[str, Any]],
+    date_ranges: List[Dict[str, str]] = None,
+    funnel_breakdown: Dict[str, str] = None,
+    funnel_next_action: Dict[str, str] = None,
+    segments: List[Dict[str, Any]] = None,
+    return_property_quota: bool = False,
+) -> Dict[str, Any]:
+    """Run a Google Analytics Data API funnel report using the v1alpha API.
+    
+    The runFunnelReport method is currently in alpha and allows you to create
+    funnel reports showing how users progress through a sequence of steps.
+    
+    Args:
+        property_id: The Google Analytics property ID. Accepted formats are:
+          - A number
+          - A string consisting of 'properties/' followed by a number
+        funnel_steps: A list of funnel steps. Each step should be a dictionary
+          containing:
+          - 'name': (str) Display name for the step
+          - 'filter_expression': (Dict) Complete filter expression for the step
+          OR for simple event-based steps:
+          - 'name': (str) Display name for the step  
+          - 'event': (str) Event name to filter on
+          Example:
+          [
+              {
+                  "name": "Page View",
+                  "filter_expression": {
+                      "funnel_field_filter": {
+                          "field_name": "eventName",
+                          "string_filter": {
+                              "match_type": "EXACT",
+                              "value": "page_view"
+                          }
+                      }
+                  }
+              },
+              {
+                  "name": "Sign Up",
+                  "event": "sign_up"
+              }
+          ]
+        date_ranges: A list of date ranges. If not provided, defaults to last 30 days.
+          Each date range should have 'start_date' and 'end_date' keys.
+          Example: [{"start_date": "2024-01-01", "end_date": "2024-01-31"}]
+        funnel_breakdown: Optional breakdown dimension to segment the funnel.
+          This creates separate funnel results for each value of the dimension.
+          Example: {"breakdown_dimension": "deviceCategory"}
+        funnel_next_action: Optional next action analysis configuration.
+          This analyzes what users do after completing or dropping off from the funnel.
+          Example: {"next_action_dimension": "eventName", "limit": 5}
+        segments: Optional list of segments to apply to the funnel.
+        return_property_quota: Whether to return current property quota information.
+        
+    Returns:
+        Dict containing the funnel report response with funnel results including:
+        - funnel_table: Table showing progression through funnel steps
+        - funnel_visualization: Data for visualizing the funnel
+        - property_quota: (if requested) Current quota usage information
+        
+    Raises:
+        ValueError: If funnel_steps is empty or contains invalid configurations
+        Exception: If the API request fails
+        
+    Example:
+        # Simple event-based funnel
+        result = await run_funnel_report(
+            property_id="123456789",
+            funnel_steps=[
+                {"name": "Landing Page", "event": "page_view"},
+                {"name": "Add to Cart", "event": "add_to_cart"},
+                {"name": "Purchase", "event": "purchase"}
+            ]
+        )
+        
+        # Advanced funnel with page path filtering
+        result = await run_funnel_report(
+            property_id="123456789",
+            funnel_steps=[
+                {
+                    "name": "Home Page View",
+                    "filter_expression": {
+                        "and_group": {
+                            "expressions": [
+                                {
+                                    "funnel_field_filter": {
+                                        "field_name": "eventName",
+                                        "string_filter": {"match_type": "EXACT", "value": "page_view"}
+                                    }
+                                },
+                                {
+                                    "funnel_field_filter": {
+                                        "field_name": "pagePath",
+                                        "string_filter": {"match_type": "EXACT", "value": "/"}
+                                    }
+                                }
+                            ]
+                        }
+                    }
+                },
+                {"name": "Purchase", "event": "purchase"}
+            ],
+            funnel_breakdown={"breakdown_dimension": "deviceCategory"},
+            date_ranges=[{"start_date": "7daysAgo", "end_date": "today"}]
+        )
+    """
+    # Validate inputs
+    if not funnel_steps:
+        raise ValueError("funnel_steps cannot be empty")
+    
+    if len(funnel_steps) < 2:
+        raise ValueError("funnel_steps must contain at least 2 steps")
+    
+    # Set default date range if not provided
+    if not date_ranges:
+        date_ranges = [{"start_date": "30daysAgo", "end_date": "today"}]
+    
+    # Validate and create funnel steps
+    steps = []
+    for i, step in enumerate(funnel_steps):
+        if not isinstance(step, dict):
+            raise ValueError(f"Step {i+1} must be a dictionary")
+        
+        step_name = step.get('name', f'Step {i+1}')
+        
+        # Build filter expression
+        if 'filter_expression' in step:
+            # Use provided filter expression
+            filter_expr = data_v1alpha.FunnelFilterExpression(step['filter_expression'])
+        elif 'event' in step:
+            # Simple event-based filter
+            filter_expr = data_v1alpha.FunnelFilterExpression(
+                funnel_event_filter=data_v1alpha.FunnelEventFilter(
+                    event_name=step['event']
+                )
+            )
+        else:
+            raise ValueError(
+                f"Step {i+1} must contain either 'filter_expression' or 'event' key"
+            )
+        
+        funnel_step = data_v1alpha.FunnelStep(
+            name=step_name,
+            filter_expression=filter_expr
+        )
+        steps.append(funnel_step)
+    
+    # Create the funnel configuration
+    funnel_config = data_v1alpha.Funnel(steps=steps)
+
+    # Create date ranges
+    date_range_objects = []
+    for dr in date_ranges:
+        if not isinstance(dr, dict) or 'start_date' not in dr or 'end_date' not in dr:
+            raise ValueError(
+                "Each date range must be a dictionary with 'start_date' and 'end_date' keys"
+            )
+        date_range_objects.append(
+            data_v1alpha.DateRange(start_date=dr['start_date'], end_date=dr['end_date'])
+        )
+
+    # Create the request
+    request = data_v1alpha.RunFunnelReportRequest(
+        property=construct_property_rn(property_id),
+        funnel=funnel_config,
+        date_ranges=date_range_objects,
+        return_property_quota=return_property_quota
+    )
+    
+    # Add breakdown if specified (this goes on the request, not the funnel)
+    if funnel_breakdown and 'breakdown_dimension' in funnel_breakdown:
+        request.funnel_breakdown = data_v1alpha.FunnelBreakdown(
+            breakdown_dimension=data_v1alpha.Dimension(
+                name=funnel_breakdown['breakdown_dimension']
+            )
+        )
+    
+    # Add next action if specified (this also goes on the request, not the funnel)
+    if funnel_next_action and 'next_action_dimension' in funnel_next_action:
+        next_action_config = data_v1alpha.FunnelNextAction(
+            next_action_dimension=data_v1alpha.Dimension(
+                name=funnel_next_action['next_action_dimension']
+            )
+        )
+        if 'limit' in funnel_next_action:
+            next_action_config.limit = funnel_next_action['limit']
+        request.funnel_next_action = next_action_config
+    
+    # Add segments if provided
+    if segments:
+        request.segments = [data_v1alpha.Segment(segment) for segment in segments]
+    
+    # Execute the request with enhanced error handling
+    try:
+        client = create_data_api_alpha_client()
+        response = await client.run_funnel_report(request=request)
+        return proto_to_dict(response)
+    except Exception as e:
+        error_msg = str(e)
+        if "INVALID_ARGUMENT" in error_msg:
+            raise ValueError(f"Invalid funnel configuration: {error_msg}")
+        elif "PERMISSION_DENIED" in error_msg:
+            raise PermissionError(f"Permission denied accessing property: {error_msg}")
+        elif "NOT_FOUND" in error_msg:
+            raise ValueError(f"Property not found: {error_msg}")
+        elif "QUOTA_EXCEEDED" in error_msg:
+            raise RuntimeError(f"API quota exceeded: {error_msg}")
+        else:
+            raise Exception(f"Failed to run funnel report: {error_msg}")
+
 
 # The `run_report` tool requires a more complex description that's generated at
 # runtime. Uses the `add_tool` method instead of an annnotation since `add_tool`
@@ -182,3 +501,9 @@ async def run_report(
     title="Run a Google Analytics Data API report using the Data API",
     description=_run_report_description(),
 )
+
+mcp.add_tool(
+    run_funnel_report,
+    title="Run a Google Analytics Data API funnel report using the Data API",
+    description=_run_funnel_report_description(),
+)