Wakapi MCP Server

Instructions

Implementation Reference

• src/mcp_tools/recent_logs.py:10-90 (handler)

  The core handler function for the 'get_recent_logs' tool. Decorated with @app.tool, it fetches heartbeats from the Wakapi API for the last N days, optionally filtered by project, sorts them by time descending, limits the results, and returns a list of formatted dictionaries.
  @app.tool async def get_recent_logs( user: str = "current", project_name: Optional[str] = None, days: int = 7, limit: int = 1000, ) -> list[dict[str, Any]]: """Get heartbeats of user for recent days (extension of heartbeats GET). Mimics https://wakatime.com/api/v1/users/{user}/heartbeats for multiple days. Requires ApiKeyAuth: Set header `Authorization` to your API Key encoded as Base64 and prefixed with `Basic`. Args: user (str, required, default="current"): Username (or current). project_name (str, optional): Filter by project. days (int, default=7): Number of days to retrieve. limit (int, default=1000): Maximum number of heartbeats. Returns: List of HeartbeatEntry: Each with id (str), project (str), language (str), entity (str), time (number), is_write (bool), branch (str), category (str), cursorpos (int), line_additions (int), line_deletions (int), lineno (int), lines (int), type (str), user_agent_id (str), user_id (str), machine_name_id (str), created_at (str). Sorted by time descending. """ client = get_wakapi_client() end_date = datetime.now() all_logs = [] current_date = end_date.date() for i in range(days): day_date = current_date - timedelta(days=i) try: day_logs = await client.get_heartbeats( user=user, date=day_date, project=project_name, limit=limit // days + 1 if days > 0 else limit, ) all_logs.extend(day_logs.data if hasattr(day_logs, "data") else day_logs) except Exception as e: raise ValueError(f"Failed to fetch recent logs: {e}") from e # Sort by time descending all_logs.sort(key=lambda log: log.time, reverse=True) # Limit total results all_logs = all_logs[:limit] # Convert to dict return [ { "id": log.id, "project": log.project, "language": log.language, "entity": log.entity, "time": log.time.timestamp() if isinstance(log.time, datetime) else log.time, "is_write": log.is_write, "branch": log.branch, "category": getattr(log, "category", None), "cursorpos": getattr(log, "cursorpos", None), "line_additions": getattr(log, "line_additions", None), "line_deletions": getattr(log, "line_deletions", None), "lineno": getattr(log, "lineno", None), "lines": getattr(log, "lines", None), "type": log.type, "user_agent_id": getattr(log, "user_agent_id", None), "user_id": log.user_id, "machine_name_id": getattr(log, "machine_name_id", None), "created_at": log.time.isoformat() if isinstance(log.time, datetime) else None, } for log in all_logs ]
• main.py:126-155 (registration)

  The initialize_tools function imports get_recent_logs and assigns it to _ , triggering the @app.tool decorator to register the tool with the MCP server.
  def initialize_tools(): """Initialize and register all Wakapi tools.""" try: # Direct imports to avoid circular import in __init__.py from mcp_tools.stats import get_stats _ = get_stats # Trigger registration from mcp_tools.projects import get_projects _ = get_projects # Trigger registration from mcp_tools.leaders import get_leaders _ = get_leaders # Trigger registration from mcp_tools.users import get_user _ = get_user # Trigger registration from mcp_tools.all_time import get_all_time_since_today _ = get_all_time_since_today # Trigger registration from mcp_tools.project_detail import get_project_detail _ = get_project_detail # Trigger registration from mcp_tools.recent_logs import get_recent_logs _ = get_recent_logs # Trigger registration from mcp_tools.connection import test_connection _ = test_connection # Trigger registration print("All Wakapi tools registered successfully.", file=sys.stderr) except ImportError as e: