pyp6xer_load_file

Instructions

Implementation Reference

• server.py:325-366 (handler)

  The pyp6xer_load_file tool function definition (handler). It loads a Primavera P6 XER file from a local path, HTTP/HTTPS URL, or base64-encoded content, parses it, stores it in the shared cache, and returns a summary of the loaded data.

  @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False))
  def pyp6xer_load_file(
      cache_key: Annotated[str, Field(description="Cache key identifying the loaded XER file (set when calling pyp6xer_load_file)")] = "default",
      file_path: Annotated[str | None, Field(description="Local file path or HTTP/HTTPS URL to the XER file")] = None,
      file_content: Annotated[str | None, Field(description="Base64-encoded XER file bytes (for direct uploads from Claude/ChatGPT)")] = None,
      ctx: Context = None,
  ) -> str:
      """Load a Primavera P6 XER file into the analysis cache.
  
      Accepts a local file path, an HTTP/HTTPS URL, or a base64-encoded string
      of the file's binary content. The loaded data is stored under cache_key
      so multiple schedules can be open simultaneously.
  
      Args:
          cache_key: Identifier for this file in the cache (default: "default").
          file_path: Local path (e.g. "/data/project.xer") or URL
                     (e.g. "https://example.com/project.xer").
          file_content: Base64-encoded XER file bytes (for direct uploads).
      """
      source, xer, raw_text = _load_xer_content(file_path, file_content)
      header, table_order, raw_tables = _parse_raw_tables(raw_text)
  
      cache = ctx.lifespan_context["cache"]
      cache[cache_key] = {
          "xer": xer,
          "raw_tables": raw_tables,
          "table_order": table_order,
          "header": header,
          "source": source,
      }
  
      proj_names = [f"{p.short_name} – {p.name}" for p in xer.projects.values()]
      result = {
          "status": "loaded",
          "cache_key": cache_key,
          "source": source,
          "projects": proj_names,
          "total_activities": len(xer.tasks),
          "total_relationships": len(xer.relationships),
          "total_resources": len(xer.resources),
      }
      return json.dumps(result, indent=2)

• server.py:299-300 (registration)

  The @mcp.tool decorator that registers pyp6xer_load_file as an MCP tool with annotations marking it as read-only, non-destructive, idempotent, and not open-world.

  @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False))
  def pyp6xer_get_activity_schema() -> str:

• server.py:270-292 (helper)

  The _load_xer_content helper function used by pyp6xer_load_file to handle decoding from local path, URL, or base64 content.

  def _load_xer_content(file_path: str | None, file_content: str | None) -> tuple[str, Xer]:
      """Load XER from path/URL/base64. Returns (source_label, Xer)."""
      if file_content:
          raw_bytes = base64.b64decode(file_content)
          text = raw_bytes.decode(Xer.CODEC, errors="replace")
          xer = Xer(text)
          return "base64_upload", xer, text
  
      if not file_path:
          raise ValueError("Provide either file_path or file_content.")
  
      if file_path.startswith(("http://", "https://")):
          response = httpx.get(file_path, timeout=60, follow_redirects=True)
          response.raise_for_status()
          text = response.content.decode(Xer.CODEC, errors="replace")
          xer = Xer(text)
          return file_path, xer, text
  
      with open(file_path, "rb") as f:
          raw_bytes = f.read()
      text = raw_bytes.decode(Xer.CODEC, errors="replace")
      xer = Xer(text)
      return file_path, xer, text

• server.py:214-249 (helper)

  The _parse_raw_tables helper function that parses raw XER table data preserving column order for round-trip write support.

  def _parse_raw_tables(content: str) -> tuple[str, list[str], dict]:
      """
      Parse XER content, preserving column order per table for round-trip write support.
  
      Returns:
          header      - ERMHDR line
          table_order - ordered list of table names
          raw_tables  - {table_name: {"cols": [...], "rows": [dict]}}
      """
      sections = content.split("%T\t")
      header = sections.pop(0).strip()
      table_order: list[str] = []
      raw_tables: dict = {}
  
      for section in sections:
          lines = [ln for ln in section.splitlines() if ln.strip()]
          if not lines:
              continue
          name = lines[0].strip()
          if len(lines) < 2:
              table_order.append(name)
              raw_tables[name] = {"cols": [], "rows": []}
              continue
          cols = lines[1].strip().split("\t")[1:]  # skip %F prefix
          rows = []
          for line in lines[2:]:
              if line.startswith("%R"):
                  vals = line.strip().split("\t")[1:]
                  # Pad if fewer values than columns
                  while len(vals) < len(cols):
                      vals.append("")
                  rows.append(dict(zip(cols, vals)))
          table_order.append(name)
          raw_tables[name] = {"cols": cols, "rows": rows}
  
      return header, table_order, raw_tables