from __future__ import annotations from typing import Any from mcp.server.fastmcp import FastMCP from app.analytics.schemas import AnalyticsQuery, AnalyticsSyncRequest from app.mcp.registry import MCPRegistry from app.security.context import auth_context def register_analytics_tools(server: FastMCP[Any], registry: MCPRegistry) -> None: """Register narrow analytics tools over the shared domain service.""" def identity() -> tuple[str, str]: context = auth_context.get() return ( context.workspace_id or "" if context else "", context.user_id or "" if context else "", ) @server.tool( name="analytics.overview", description="Get workspace or project analytics totals." ) async def analytics_overview(filters: dict[str, Any] | None = None) -> dict[str, Any]: async def action() -> dict[str, Any]: workspace_id, _ = identity() result = await registry.container.analytics.overview( workspace_id, AnalyticsQuery.model_validate(filters or {}) ) return result.model_dump(mode="json") return await registry.run_metadata_tool( "analytics.overview", action, required_scope="analytics:read" ) @server.tool( name="analytics.timeseries", description="Get timezone-safe analytics time series." ) async def analytics_timeseries(filters: dict[str, Any] | None = None) -> dict[str, Any]: async def action() -> dict[str, Any]: workspace_id, _ = identity() result = await registry.container.analytics.timeseries( workspace_id, AnalyticsQuery.model_validate(filters or {}) ) return result.model_dump(mode="json") return await registry.run_metadata_tool( "analytics.timeseries", action, required_scope="analytics:read" ) @server.tool(name="analytics.platforms", description="Compare available provider metrics.") async def analytics_platforms(filters: dict[str, Any] | None = None) -> dict[str, Any]: return await analytics_overview(filters) @server.tool( name="analytics.top_content", description="List top content by an available metric." ) async def analytics_top_content(filters: dict[str, Any] | None = None) -> dict[str, Any]: async def action() -> dict[str, Any]: workspace_id, _ = identity() query = AnalyticsQuery.model_validate({**(filters or {}), "limit": 10}) result = await registry.container.analytics.posts(workspace_id, query) return result.model_dump(mode="json") return await registry.run_metadata_tool( "analytics.top_content", action, required_scope="analytics:read" ) @server.tool(name="analytics.post", description="Get analytics for one MediaRouter post.") async def analytics_post(post_id: str, filters: dict[str, Any] | None = None) -> dict[str, Any]: async def action() -> dict[str, Any]: workspace_id, _ = identity() result = await registry.container.analytics.post( workspace_id, post_id, AnalyticsQuery.model_validate(filters or {}) ) return result.model_dump(mode="json") return await registry.run_metadata_tool( "analytics.post", action, required_scope="analytics:read" ) @server.tool( name="analytics.sync", description="Queue an idempotent analytics synchronization." ) async def analytics_sync(payload: dict[str, Any]) -> dict[str, Any]: async def action() -> dict[str, Any]: workspace_id, user_id = identity() result = await registry.container.analytics.create_sync( workspace_id, user_id, AnalyticsSyncRequest.model_validate(payload) ) return result.model_dump(mode="json") return await registry.run_metadata_tool( "analytics.sync", action, required_scope="analytics:sync" ) @server.tool(name="analytics.sync_status", description="Get one analytics sync status.") async def analytics_sync_status(sync_run_id: str) -> dict[str, Any]: async def action() -> dict[str, Any]: workspace_id, _ = identity() run = await registry.container.analytics.repository.get_sync(workspace_id, sync_run_id) return registry.container.analytics._sync_view(run).model_dump(mode="json") return await registry.run_metadata_tool( "analytics.sync_status", action, required_scope="analytics:read" )