Productive Simple MCP

Instructions

Implementation Reference

• server.py:121-186 (registration)

  MCP tool registration (@mcp.tool) with full input schema defined via Annotated parameters and Field descriptions. Includes comprehensive docstring with usage examples. Thin wrapper delegating to tools.py implementation.

  async def get_recent_activity(
      ctx: Context,
      hours: Annotated[
          int,
          Field(
              description="Number of hours to look back (default: 24, use 168 for a week)"
          ),
      ] = 24,
      user_id: Annotated[
          int, Field(description="Optional: Filter by specific user/person ID")
      ] = None,
      project_id: Annotated[
          int, Field(description="Optional: Filter by specific project ID")
      ] = None,
      activity_type: Annotated[
          int,
          Field(
              description="Optional: Filter by activity type (1: Comment, 2: Changeset, 3: Email)"
          ),
      ] = None,
      item_type: Annotated[
          str,
          Field(
              description="Optional: Filter by item type. Accepted values include: Task, Page, Project, Person, Discussion, TimeEntry, Section, TaskList, Dashboard, Team. Note: This list is not exhaustive."
          ),
      ] = None,
      event_type: Annotated[
          str,
          Field(
              description="Optional: Filter by event type. Common values include: create, copy, edit, delete. Note: Use get_tasks with filter[status][eq]=2 to find closed tasks."
          ),
      ] = None,
      task_id: Annotated[
          int, Field(description="Optional: Filter by specific task ID")
      ] = None,
      max_results: Annotated[
          int,
          Field(description="Optional maximum number of activities to return (max: 200)"),
      ] = None,
  ) -> Dict[str, Any]:
      """Get a summarized feed of recent activities and updates.
  
      Returns recent changes, task updates, comments, new documents and activities in chronological order.
  
      Examples:
          get_recent_activity()  # Last 24 hours, all activity
          get_recent_activity(hours=168)  # Last week
          get_recent_activity(hours=48, project_id=343136)  # Last 2 days on specific project
          get_recent_activity(hours=24, user_id=12345)  # What a specific user did today
          get_recent_activity(hours=24, activity_type=1)  # Only comments from last day
          get_recent_activity(hours=168, item_type='Task')  # Task activities from last week
          get_recent_activity(hours=168, event_type='edit')  # Task edits from last week
          get_tasks(extra_filters={'filter[status][eq]': 2}, sort='-updated_at', page_size=10)  # Recently closed tasks
      """
      return await tools.get_recent_activity(
          ctx,
          hours=hours,
          user_id=user_id,
          project_id=project_id,
          activity_type=activity_type,
          item_type=item_type,
          event_type=event_type,
          task_id=task_id,
          max_results=max_results,
      )

• tools.py:354-451 (handler)

  Primary handler function implementing the core logic: calculates time filter, builds API params for client.get_activities(), handles empty results, filters response, adds summary metadata using helpers, and error handling.

  async def get_recent_activity(
      ctx: Context,
      hours: int = 24,
      user_id: int = None,
      project_id: int = None,
      activity_type: int = None,
      item_type: str = None,
      event_type: str = None,
      task_id: int = None,
      max_results: int = None
  ) -> ToolResult:
      """Summarize recent activities within a time window.
  
      Developer notes:
      - Builds filter[after] from UTC now minus `hours`.
      - Optional filters map directly: person_id, project_id, type (1:Comment,2:Changeset,3:Email), item_type, event, task_id.
      - Respects API page[size] limit (<=200) via max_results.
      - Response is sanitized and meta is enriched with basic counts via _summarize_activities.
      - Avoids unsupported sorts on /activities.
      """
      try:
          from datetime import datetime, timedelta
  
          if max_results is None:
              max_results = config.items_per_page
  
          # Validate max_results
          if max_results > 200:
              await ctx.warning("max_results exceeds API limit of 200, using 200")
              max_results = 200
          
          # Calculate the cutoff time
          cutoff_time = datetime.utcnow() - timedelta(hours=hours)
          after_date = cutoff_time.isoformat() + "Z"
          
          await ctx.info(f"Fetching activities from the last {hours} hours")
          
          # Build comprehensive filter params
          params = {
              "filter[after]": after_date,
              "page[size]": max_results
          }
          
          # Apply optional filters
          if user_id:
              params["filter[person_id]"] = user_id
              
          if project_id:
              params["filter[project_id]"] = project_id
              
          if activity_type:
              params["filter[type]"] = activity_type
              
          if item_type:
              params["filter[item_type]"] = item_type
              
          if event_type:
              params["filter[event]"] = event_type
              
          if task_id:
              params["filter[task_id]"] = task_id
          
          result = await client.get_activities(params=params)
          
          if not result.get("data") or len(result["data"]) == 0:
              await ctx.info("No recent activities found")
              return {
                  "data": [],
                  "meta": {
                      "message": f"No activities found in the last {hours} hours",
                      "hours": hours,
                      "filters_applied": _get_applied_filters(params),
                      "cutoff_time": after_date
                  }
              }
          
          filtered = filter_response(result)
          
          # Enhance metadata with activity summary
          activity_summary = _summarize_activities(filtered.get("data", []))
          filtered["meta"] = filtered.get("meta", {})
          filtered["meta"].update({
              "activity_summary": activity_summary,
              "total_activities": len(filtered.get("data", [])),
              "filters_applied": _get_applied_filters(params),
              "cutoff_time": after_date
          })
          
          await ctx.info(f"Successfully retrieved {len(result['data'])} recent activities")
          
          return filtered
          
      except ProductiveAPIError as e:
          await _handle_productive_api_error(ctx, e, "activities")
      except Exception as e:
          await ctx.error(f"Unexpected error fetching recent updates: {str(e)}")
          raise e

• tools.py:453-466 (helper)

  Helper function to extract and format applied filter parameters for metadata logging in get_recent_activity.

  def _get_applied_filters(params: dict) -> dict:
      """Extract and format the filters that were actually applied."""
      applied_filters = {}
      
      # Remove pagination and standard params
      filter_params = {k: v for k, v in params.items() if k.startswith("filter[")}
      
      for key, value in filter_params.items():
          # Extract filter name from key like "filter[person_id]"
          filter_name = key.replace("filter[", "").replace("]", "")
          applied_filters[filter_name] = value
      
      return applied_filters

• tools.py:581-612 (helper)

  Helper function to generate summary statistics of activities grouped by type, event, and item_type. Used to enrich the meta response in get_recent_activity.

  def _summarize_activities(activities: list) -> dict:
      """Create a summary of activities by type and event."""
      summary = {
          "by_type": {},
          "by_event": {},
          "by_item_type": {},
          "total": len(activities)
      }
      
      for activity in activities:
          if not isinstance(activity, dict):
              continue
              
          attributes = activity.get("attributes", {})
          activity_type = attributes.get("type")
          event_type = attributes.get("event")
          item_type = attributes.get("item_type")
          
          # Count by activity type
          if activity_type:
              summary["by_type"][activity_type] = summary["by_type"].get(activity_type, 0) + 1
              
          # Count by event type
          if event_type:
              summary["by_event"][event_type] = summary["by_event"].get(event_type, 0) + 1
              
          # Count by item type
          if item_type:
              summary["by_item_type"][item_type] = summary["by_item_type"].get(item_type, 0) + 1
      
      return summary

• server.py:123-160 (schema)

  Input schema definitions using Pydantic Annotated and Field for parameter validation, descriptions, and types in the MCP tool.

      hours: Annotated[
          int,
          Field(
              description="Number of hours to look back (default: 24, use 168 for a week)"
          ),
      ] = 24,
      user_id: Annotated[
          int, Field(description="Optional: Filter by specific user/person ID")
      ] = None,
      project_id: Annotated[
          int, Field(description="Optional: Filter by specific project ID")
      ] = None,
      activity_type: Annotated[
          int,
          Field(
              description="Optional: Filter by activity type (1: Comment, 2: Changeset, 3: Email)"
          ),
      ] = None,
      item_type: Annotated[
          str,
          Field(
              description="Optional: Filter by item type. Accepted values include: Task, Page, Project, Person, Discussion, TimeEntry, Section, TaskList, Dashboard, Team. Note: This list is not exhaustive."
          ),
      ] = None,
      event_type: Annotated[
          str,
          Field(
              description="Optional: Filter by event type. Common values include: create, copy, edit, delete. Note: Use get_tasks with filter[status][eq]=2 to find closed tasks."
          ),
      ] = None,
      task_id: Annotated[
          int, Field(description="Optional: Filter by specific task ID")
      ] = None,
      max_results: Annotated[
          int,
          Field(description="Optional maximum number of activities to return (max: 200)"),
      ] = None,
  ) -> Dict[str, Any]: