analytics

Instructions

Implementation Reference

• tools/analytics.py:307-327 (handler)

  Async handler for the analytics tool. Dispatches to summary_report(), revenue_projection(), or returns current/pricing data based on report_type argument.

  async def analytics_handler(arguments: dict) -> dict:
      engine = get_engine()
      report_type = arguments.get("report_type", "summary")
  
      if report_type == "summary":
          return engine.summary_report()
      elif report_type == "revenue_projection":
          days = int(arguments.get("projection_days", 30))
          return engine.revenue_projection(days)
      elif report_type == "pricing":
          return {
              "current_prices": {
                  name: stats.current_price
                  for name, stats in engine._stats.items()
              },
              "price_history": {
                  name: stats.price_history[-3:]
                  for name, stats in engine._stats.items()
              },
          }
      return {"error": f"Unknown report type: {report_type}"}

• tools/analytics.py:280-295 (schema)

  Input schema for the analytics tool: accepts 'report_type' (enum: summary, revenue_projection, pricing) and optional 'projection_days' integer.

  ANALYTICS_TOOL_SCHEMA = {
      "type": "object",
      "properties": {
          "report_type": {
              "type": "string",
              "enum": ["summary", "revenue_projection", "pricing"],
              "description": "Type of analytics report to return",
              "default": "summary",
          },
          "projection_days": {
              "type": "integer",
              "description": "Days to project revenue forward (for revenue_projection)",
              "default": 30,
          },
      },
  }

• tools/analytics.py:330-344 (registration)

  Registration function that registers the 'analytics' tool with name, description, input_schema, price, stripe_price_id, handler, and category='internal'.

  def register(registry) -> None:
      from server import ToolDefinition
      registry.register(ToolDefinition(
          name="analytics",
          description=(
              "Query usage analytics and revenue data for this MCP server. "
              "Returns call volumes, revenue, success rates, latency, "
              "and dynamic pricing recommendations."
          ),
          input_schema=ANALYTICS_TOOL_SCHEMA,
          price_per_call_usd=0.001,
          stripe_price_id=os.getenv("STRIPE_PRICE_ANALYTICS", "price_demo_analytics"),
          handler=analytics_handler,
          category="internal",
      ))

• tools/analytics.py:97-105 (helper)

  AnalyticsEngine class that tracks per-tool call volumes, latency, error rates, revenue, and runs periodic pricing optimisation.

  class AnalyticsEngine:
      """
      Tracks all tool calls and runs periodic pricing optimisation.
      Integrates with the MCP server's UsageTracker.
      """
  
      def __init__(self):
          self._stats: dict[str, ToolStats] = {}
          self._last_optimised: float = 0.0

• tools/analytics.py:53-90 (helper)

  ToolStats dataclass capturing per-tool metrics: calls, revenue, latency, customers, hourly call distribution, price history.

  @dataclass
  class ToolStats:
      tool_name: str
      total_calls: int = 0
      successful_calls: int = 0
      failed_calls: int = 0
      total_revenue_usd: float = 0.0
      total_latency_ms: float = 0.0
      unique_customers: set = field(default_factory=set)
      hourly_calls: dict = field(default_factory=lambda: defaultdict(int))
      current_price: float = 0.0
      price_history: list = field(default_factory=list)
  
      @property
      def success_rate(self) -> float:
          if self.total_calls == 0:
              return 1.0
          return self.successful_calls / self.total_calls
  
      @property
      def avg_latency_ms(self) -> float:
          if self.successful_calls == 0:
              return 0.0
          return self.total_latency_ms / self.successful_calls
  
      @property
      def revenue_per_call(self) -> float:
          if self.successful_calls == 0:
              return 0.0
          return self.total_revenue_usd / self.successful_calls
  
      def calls_in_last_n_hours(self, n: int = 24) -> int:
          now_hour = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
          total = 0
          for i in range(n):
              hour_key = (now_hour - timedelta(hours=i)).isoformat()
              total += self.hourly_calls.get(hour_key, 0)
          return total