MCP Logger

Instructions

Implementation Reference

• src/main.py:727-800 (handler)

  The implementation of the `search_logs` MCP tool, which performs SQL queries against the 'workouts', 'nutrition_days', and 'body_metrics' tables to search for notes or tags.

  def search_logs(
      query: str,
      domains: Optional[list[str]] = None,
      from_date: Optional[str] = None,
      to_date: Optional[str] = None,
      limit: int = 20,
  ) -> dict[str, list[dict[str, Any]]]:
      """Search across workouts, nutrition days, and body metrics."""
      domains = domains or ["workout", "nutrition", "body"]
      results = []
  
      conn = get_connection()
      cursor = conn.cursor()
  
      if "workout" in domains:
          filters = ["(notes LIKE ? OR workout_type LIKE ? OR tags LIKE ?)"]
          params = [f"%{query}%", f"%{query}%", f"%{query}%"]
          if from_date:
              filters.append("date_time >= ?")
              params.append(f"{_ensure_date(from_date)}T00:00:00")
          if to_date:
              filters.append("date_time <= ?")
              params.append(f"{_ensure_date(to_date)}T23:59:59")
          base = "SELECT * FROM workouts WHERE " + " AND ".join(filters) + " LIMIT ?"
          params.append(limit)
          cursor.execute(base, params)
          for row in cursor.fetchall():
              workout = _hydrate_workout(conn, _row_to_dict(row))
              results.append({"domain": "workout", "workout": workout})
  
      if len(results) >= limit:
          conn.close()
          return {"results": results[:limit]}
  
      remaining = limit - len(results)
  
      if "nutrition" in domains:
          filters = ["(notes LIKE ?)"]
          params = [f"%{query}%"]
          if from_date:
              filters.append("date >= ?")
              params.append(_ensure_date(from_date))
          if to_date:
              filters.append("date <= ?")
              params.append(_ensure_date(to_date))
          base = "SELECT * FROM nutrition_days WHERE " + " AND ".join(filters) + " LIMIT ?"
          params.append(remaining)
          cursor.execute(base, params)
          for row in cursor.fetchall():
              results.append({"domain": "nutrition", "nutrition": _row_to_dict(row)})
  
      if len(results) >= limit:
          conn.close()
          return {"results": results[:limit]}
  
      remaining = limit - len(results)
  
      if "body" in domains:
          filters = ["(notes LIKE ?)"]
          params = [f"%{query}%"]
          if from_date:
              filters.append("date >= ?")
              params.append(_ensure_date(from_date))
          if to_date:
              filters.append("date <= ?")
              params.append(_ensure_date(to_date))
          base = "SELECT * FROM body_metrics WHERE " + " AND ".join(filters) + " LIMIT ?"
          params.append(remaining)
          cursor.execute(base, params)
          for row in cursor.fetchall():
              results.append({"domain": "body", "body": _row_to_dict(row)})
  
      conn.close()
      return {"results": results}