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]: