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