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: