Productive Simple MCP

Instructions

Implementation Reference

• tools.py:468-578 (handler)

  Core handler implementation that aggregates task history by fetching recent activities, parsing changesets for status and assignment updates, identifying milestones from comments, and compiling a comprehensive summary.
  async def get_task_history( ctx: Context, task_id: int, hours: int = 720 # Default to 30 days for comprehensive history ) -> ToolResult: """Get comprehensive history for a specific task. Developer notes: - Aggregates historical data from activities and task events - Returns status history, assignment history, milestones, and activity summary - Ignores unavailable data gracefully (empty arrays/null values) - Uses get_recent_activity with task_id filter for historical data - Now extracts status transitions from changeset (workflow_status_id) """ try: # Get the task details first to verify it exists await ctx.info(f"Fetching history for task {task_id}") task_result = await get_task(ctx, task_id) if not task_result.get("data"): await ctx.error(f"Task {task_id} not found") return { "task_id": task_id, "error": "Task not found", "status_history": [], "assignment_history": [], "milestones": [], "activity_summary": {} } # Get recent activities for this task (comprehensive history) activity_result = await get_recent_activity( ctx, hours=hours, task_id=task_id, max_results=200 # Maximum allowed by API ) activities = activity_result.get("data", []) # Parse activities to extract status changes, assignments, milestones status_history = [] assignment_history = [] milestones = [] for activity in activities: attributes = activity.get("attributes", {}) event_type = attributes.get("event") item_type = attributes.get("item_type") created_at = attributes.get("created_at") person_name = attributes.get("person_name", "Unknown") # Status changes (workflow_status_id in changeset) changeset = attributes.get("changeset", []) if item_type and item_type.lower() == "task" and event_type in ["update", "edit"]: for change in changeset: if "workflow_status_id" in change: status_from = change["workflow_status_id"][0]["value"] if len(change["workflow_status_id"]) > 0 else None status_to = change["workflow_status_id"][1]["value"] if len(change["workflow_status_id"]) > 1 else None status_history.append({ "from": status_from, "to": status_to, "changed_at": created_at }) # Assignment changes (parse assignee in changeset) if item_type and item_type.lower() == "task" and event_type in ["update", "edit"]: for change in changeset: if "assignee" in change: # The changeset shows the new assignee only assignee_to = change["assignee"][0]["value"] if len(change["assignee"]) > 0 else None assignment_history.append({ "assigned_to": assignee_to, "changed_at": created_at }) # Milestones (comments with milestone keywords or custom fields) if item_type and item_type.lower() == "comment" and "milestone" in attributes.get("item_name", "").lower(): milestones.append({ "milestone": attributes.get("item_name", "Milestone"), "completed_at": created_at, "completed_by": person_name }) # Build activity summary activity_summary = { "total_activities": len(activities), "total_comments": len([a for a in activities if a.get("attributes", {}).get("item_type") == "Comment"]), "total_changes": len([a for a in activities if a.get("attributes", {}).get("item_type") == "Task"]), "total_status_changes": len(status_history), "total_assignments": len(assignment_history), "total_milestones": len(milestones) } # Build final history response history_response = { "task_id": task_id, "status_history": status_history, "assignment_history": assignment_history, "milestones": milestones, "activity_summary": activity_summary } await ctx.info(f"Successfully retrieved history for task {task_id}") return history_response except ProductiveAPIError as e: await _handle_productive_api_error(ctx, e, f"task history for {task_id}") except Exception as e: await ctx.error(f"Unexpected error fetching task history: {str(e)}") raise e
• server.py:271-300 (registration)

  MCP tool registration using @mcp.tool decorator. Includes input schema validation via Annotated Fields with descriptions and constraints, comprehensive docstring with examples, and delegates to the core handler in tools.py.
  @mcp.tool async def get_task_history( ctx: Context, task_id: Annotated[ int, Field(description="The unique Productive task identifier (internal ID)") ], hours: Annotated[ int, Field( description="Number of hours to look back for activity history (default: 720 = 30 days)", ge=1, le=8760 # 1 year max ) ] = 720 ) -> Dict[str, Any]: """Get comprehensive history for a specific task. Returns aggregated task history including: - Status history: Timeline of status changes with timestamps and responsible users - Assignment history: Who worked on the task and when assignments changed - Milestones: Key deliverables and completion markers from comments and activities - Activity summary: Counts of comments, changes, status updates, assignments, and milestones Examples: get_task_history(14677921) # Default 30-day history get_task_history(14677921, hours=168) # Last week only get_task_history(14677921,1 hours=24) # Last 24 hours """ return await tools.get_task_history(ctx, task_id, hours)
• server.py:274-284 (schema)

  Pydantic schema definitions for tool inputs using Annotated and Field, specifying types, descriptions, and validation constraints (hours: 1-8760). Part of the tool registration.
  task_id: Annotated[ int, Field(description="The unique Productive task identifier (internal ID)") ], hours: Annotated[ int, Field( description="Number of hours to look back for activity history (default: 720 = 30 days)", ge=1, le=8760 # 1 year max ) ] = 720