pyp6xer_get_activity
Retrieve comprehensive activity data including dates, float, costs, resources, and predecessor/successor relationships for a given task code.
Instructions
Get full details for a single activity including dates, float, costs, resources assigned, and predecessor/successor relationships.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| task_code | Yes | Activity task code (e.g. A1000) | |
| cache_key | No | Cache key identifying the loaded XER file (set when calling pyp6xer_load_file) | default |
| proj_id | No | Project ID or short name; uses first project if omitted | |
| fields | No | Subset of field names to return; call pyp6xer_get_activity_schema to see available names |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- server.py:510-595 (handler)The handler function for pyp6xer_get_activity. Looks up a task by task_code, builds a detailed dict with dates, costs, predecessors, successors, resources, and memos, optionally filters fields, and returns JSON.
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False)) def pyp6xer_get_activity( task_code: Annotated[str, Field(description="Activity task code (e.g. A1000)")], cache_key: Annotated[str, Field(description="Cache key identifying the loaded XER file (set when calling pyp6xer_load_file)")] = "default", proj_id: Annotated[str | None, Field(description="Project ID or short name; uses first project if omitted")] = None, fields: Annotated[list[str] | None, Field(description="Subset of field names to return; call pyp6xer_get_activity_schema to see available names")] = None, ctx: Context = None, ) -> str: """Get full details for a single activity including dates, float, costs, resources assigned, and predecessor/successor relationships. Args: task_code: Activity ID (e.g. 'A1000'). cache_key: Cache key of the loaded file. proj_id: Optional project filter. fields: Subset of fields to return. Call pyp6xer_get_activity_schema for available names. Omit to return all fields. """ xer = _get_xer(ctx, cache_key) tasks = _get_tasks(xer, proj_id) task = next((t for t in tasks if t.task_code == task_code), None) if task is None: raise ValueError(f"Activity '{task_code}' not found.") # Build full detail dict (no projection yet — detail fields extend summary) detail = _task_to_dict(task) detail.update({ "actual_start": _fmt_date(task.act_start_date), "actual_finish": _fmt_date(task.act_end_date), "early_start": _fmt_date(task.early_start_date), "early_finish": _fmt_date(task.early_end_date), "late_start": _fmt_date(task.late_start_date), "late_finish": _fmt_date(task.late_end_date), "expected_finish": _fmt_date(task.expect_end_date), "duration_type": task.duration_type, "percent_complete_type": task.complete_pct_type, "physical_pct": round(task.phys_complete_pct * 100, 1), "float_path": task.float_path, "constraints": { "primary": { "type": task.cstr_type, "date": _fmt_date(task.cstr_date), }, "secondary": { "type": task.cstr_type2, "date": _fmt_date(task.cstr_date2), }, }, "at_completion_cost": task.at_completion_cost, "cost_variance": task.cost_variance, "predecessors": [ { "task_code": lnk.task.task_code, "name": lnk.task.name, "type": lnk.link, "lag_days": lnk.lag, } for lnk in task.predecessors ], "successors": [ { "task_code": lnk.task.task_code, "name": lnk.task.name, "type": lnk.link, "lag_days": lnk.lag, } for lnk in task.successors ], "resources": [ { "resource": res.resource.name, "type": res.rsrc_type, "target_qty": res.target_qty, "actual_qty": res.act_reg_qty + res.act_ot_qty, "remain_qty": res.remain_qty, "target_cost": res.target_cost, "actual_cost": res.act_total_cost, "remain_cost": res.remain_cost, } for res in task.resources.values() ], "memos": [m.memo_text for m in task.memos] if task.memos else [], }) if fields: detail = {k: v for k, v in detail.items() if k in fields} return json.dumps(detail, indent=2) - server.py:121-126 (schema)Schema definition listing the additional detail fields available on get_activity (beyond the summary fields).
ACTIVITY_DETAIL_FIELDS: list[str] = ACTIVITY_SUMMARY_FIELDS + [ "actual_start", "actual_finish", "early_start", "early_finish", "late_start", "late_finish", "resources", "predecessors", "successors", ] - server.py:510-510 (registration)Registration of the tool via the @mcp.tool decorator on FastMCP instance.
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False)) - server.py:175-211 (helper)Helper function that converts a task object to a standard dictionary with summary fields, used by get_activity as the base before adding detail fields.
def _task_to_dict(task, fields: list[str] | None = None) -> dict: """Standard activity summary dict. Pass fields to project to a subset.""" try: start = _fmt_date(task.start) except (ValueError, AttributeError): start = "" try: finish = _fmt_date(task.finish) except (ValueError, AttributeError): finish = "" full = { "task_id": task.uid, "task_code": task.task_code, "name": task.name, "status": task.status.value, "type": task.type.value, "wbs": task.wbs.full_code if task.wbs else "", "wbs_name": task.wbs.name if task.wbs else "", "start": start, "finish": finish, "target_start": _fmt_date(task.target_start_date), "target_finish": _fmt_date(task.target_end_date), "original_duration_days": task.original_duration, "remaining_duration_days": task.remaining_duration, "total_float_days": task.total_float, "free_float_days": task.free_float, "is_critical": task.is_critical, "is_longest_path": task.is_longest_path, "percent_complete": round(task.percent_complete * 100, 1), "budgeted_cost": task.budgeted_cost, "actual_cost": task.actual_cost, "remaining_cost": task.remaining_cost, } if fields: return {k: v for k, v in full.items() if k in fields} return full - server.py:168-172 (helper)Helper that retrieves tasks optionally filtered by project ID, used by get_activity to find the task.
def _get_tasks(xer: Xer, proj_id: str | None): """Return tasks for a project (or all tasks if proj_id is None).""" if proj_id: return _get_project(xer, proj_id).tasks return list(xer.tasks.values())