pyp6xer_lookahead

Instructions

Implementation Reference

• server.py:1036-1085 (handler)

  The handler function for the pyp6xer_lookahead tool. It retrieves activities active within a configurable lookahead window (default 14 days) from the project's data date. Filters out completed activities, includes in-progress and starting activities, and returns them sorted by start date.

  @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False))
  def pyp6xer_lookahead(
      cache_key: Annotated[str, Field(description="Cache key identifying the loaded XER file (set when calling pyp6xer_load_file)")] = "default",
      days_ahead: Annotated[int, Field(description="Number of calendar days ahead to include in the lookahead window", ge=1)] = 14,
      proj_id: Annotated[str | None, Field(description="Project ID or short name; uses first project if omitted")] = None,
      ctx: Context = None,
  ) -> str:
      """Return activities active within the next N days from the data date.
  
      An activity is included if: finish >= data_date AND start <= data_date + days_ahead.
      This covers in-progress activities and those starting in the window.
  
      Args:
          cache_key:  Cache key of the loaded file.
          days_ahead: Lookahead window length in days (default 14).
          proj_id:    Optional project filter.
      """
      xer = _get_xer(ctx, cache_key)
      proj = _get_project(xer, proj_id)
      tasks = proj.tasks if proj_id else list(xer.tasks.values())
  
      data_date = proj.data_date
      window_end = data_date + timedelta(days=days_ahead)
  
      activities = []
      for t in tasks:
          if t.status.is_completed:
              continue
          try:
              start = t.start
              finish = t.finish
          except Exception:
              continue
          if finish >= data_date and start <= window_end:
              d = _task_to_dict(t, fields=[
                  "task_code", "name", "start", "finish",
                  "target_finish", "total_float_days", "percent_complete",
                  "is_critical", "status", "wbs_name",
              ])
              activities.append(d)
  
      activities.sort(key=lambda a: a.get("start") or "")
  
      return json.dumps({
          "data_date": _fmt_date(data_date),
          "window_end": _fmt_date(window_end),
          "days_ahead": days_ahead,
          "activity_count": len(activities),
          "activities": activities,
      }, indent=2)

• server.py:1036-1037 (registration)

  The tool is registered via the @mcp.tool decorator on FastMCP instance, which registers it as an MCP tool named 'pyp6xer_lookahead'.

  @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False))
  def pyp6xer_lookahead(

• server.py:1038-1042 (schema)

  Input schema/type definitions for the tool's parameters: cache_key (str), days_ahead (int, ge=1, default 14), proj_id (optional str), and ctx (Context).

      cache_key: Annotated[str, Field(description="Cache key identifying the loaded XER file (set when calling pyp6xer_load_file)")] = "default",
      days_ahead: Annotated[int, Field(description="Number of calendar days ahead to include in the lookahead window", ge=1)] = 14,
      proj_id: Annotated[str | None, Field(description="Project ID or short name; uses first project if omitted")] = None,
      ctx: Context = None,
  ) -> str:

• server.py:149-151 (helper)

  Helper function used by pyp6xer_lookahead to retrieve the Xer object from the shared cache.

  def _get_xer(ctx: Context, cache_key: str) -> Xer:
      return _get_cache(ctx, cache_key)["xer"]

• server.py:129-132 (helper)

  Helper function used by pyp6xer_lookahead to format dates for the JSON output.

  def _fmt_date(dt: datetime | None) -> str:
      if dt is None:
          return ""
      return dt.strftime(DATE_FMT)