Productive Simple MCP

Instructions

Implementation Reference

• tools.py:786-904 (handler)

  Core handler function implementing the quick_search tool logic: calls the Productive API via client, post-processes results by filtering essential fields, constructs webapp URLs, and hydrates task results with workflow status details.
  async def quick_search( ctx: Context, query: str, search_types: list[str] = None, deep_search: bool = True, page: int = 1, per_page: int = 50 ) -> ToolResult: """Quick search across projects, tasks, pages, and actions. This tool provides fast, comprehensive search across all Productive content types including projects, tasks, pages, and actions. It's optimized for quick lookups and general search queries. Args: query: Search query string search_types: List of types to search (action, project, task, page). Defaults to ["action", "project", "task", "page"] if not provided. deep_search: Whether to perform deep search (default: True) page: Page number for pagination (default: 1) per_page: Results per page (default: 50) Returns: Search results from Productive API including: - Matching projects, tasks, pages, and actions - Relevance scores and metadata - Full entity details for each match """ try: # Set default search_types if not provided if search_types is None: search_types = ["action", "project", "task", "page"] await ctx.info(f"Quick search with query: '{query}'") # Call the quick search method result = await client.quick_search( query=query, search_types=search_types, deep_search=deep_search, page=page, per_page=per_page ) await ctx.info(f"Successfully retrieved {len(result.get('data', []))} search results") # Filter results to include only essential fields filtered_data = [] for item in result.get("data", []): attributes = item.get("attributes", {}) record_type = attributes.get("record_type", "") record_id = attributes.get("record_id", "") # Construct webapp URL (use raw record_type/record_id path; task hydration adds exact URL later) webapp_url = f"https://app.productive.io/27956-lineout/{record_type}s/{record_id}" filtered_item = { "record_id": record_id, "record_type": record_type, "title": attributes.get("title", ""), "subtitle": attributes.get("subtitle", ""), "icon_url": attributes.get("icon_url"), "status": attributes.get("status", ""), "project_name": attributes.get("meta", {}).get("project_name", ""), "updated_at": attributes.get("updated_at", ""), "webapp_url": webapp_url } # For tasks, hydrate with full task details to expose workflow_status_name (custom status) if record_type == "task" and record_id: try: task_details = await client.get_task(int(record_id)) filtered_task = filter_response(task_details) task_data = filtered_task.get("data", {}) if isinstance(filtered_task, dict) else {} task_attrs = task_data.get("attributes", {}) if isinstance(task_data, dict) else {} workflow_status = task_attrs.get("workflow_status_name") if workflow_status: filtered_item["workflow_status_name"] = workflow_status # Prefer canonical webapp URL from filtered task if present if "webapp_url" in task_data: filtered_item["webapp_url"] = task_data["webapp_url"] except Exception as task_error: await ctx.warning(f"Could not fetch workflow status for task {record_id}: {str(task_error)}") filtered_data.append(filtered_item) return { "data": filtered_data, "meta": { "query": query, "search_types": search_types, "deep_search": deep_search, "page": page, "per_page": per_page, "total_results": len(filtered_data) } } except ProductiveAPIError as e: await ctx.error(f"Quick search failed: {e.message}") return { "data": [], "meta": { "error": str(e), "status_code": e.status_code, "query": query } } except Exception as e: await ctx.error(f"Unexpected error during quick search: {str(e)}") return { "data": [], "meta": { "error": str(e), "query": query } }
• server.py:77-117 (registration)

  MCP tool registration using @mcp.tool decorator. Defines input schema via Annotated parameters with descriptions. Thin proxy wrapper that delegates execution to tools.quick_search.
  @mcp.tool async def quick_search( ctx: Context, query: Annotated[str, Field(description="Search query string")], search_types: Annotated[ list[str], Field( description="List of types to search (action, project, task, page). Defaults to all." ), ] = None, deep_search: Annotated[ bool, Field(description="Whether to perform deep search") ] = True, page: Annotated[int, Field(description="Page number for pagination")] = 1, per_page: Annotated[int, Field(description="Results per page")] = 50, ) -> Dict[str, Any]: """Quick search across projects, tasks, pages, and actions. This tool provides fast, comprehensive search across all Productive content types including projects, tasks, pages, and actions. It's optimized for quick lookups and general search queries. Returns: Search results from Productive API including: - Matching projects, tasks, pages, and actions - Relevance scores and metadata - Full entity details for each match Examples: quick_search("red") # Search for "red" across all content types quick_search("project", search_types=["project"]) # Search only in projects quick_search("meeting", deep_search=False) # Quick search without deep scan """ return await tools.quick_search( ctx, query=query, search_types=search_types, deep_search=deep_search, page=page, per_page=per_page, )
• server.py:78-117 (schema)

  Input schema and type definitions for the quick_search tool parameters using pydantic Annotated and Field for validation and descriptions.
  async def quick_search( ctx: Context, query: Annotated[str, Field(description="Search query string")], search_types: Annotated[ list[str], Field( description="List of types to search (action, project, task, page). Defaults to all." ), ] = None, deep_search: Annotated[ bool, Field(description="Whether to perform deep search") ] = True, page: Annotated[int, Field(description="Page number for pagination")] = 1, per_page: Annotated[int, Field(description="Results per page")] = 50, ) -> Dict[str, Any]: """Quick search across projects, tasks, pages, and actions. This tool provides fast, comprehensive search across all Productive content types including projects, tasks, pages, and actions. It's optimized for quick lookups and general search queries. Returns: Search results from Productive API including: - Matching projects, tasks, pages, and actions - Relevance scores and metadata - Full entity details for each match Examples: quick_search("red") # Search for "red" across all content types quick_search("project", search_types=["project"]) # Search only in projects quick_search("meeting", deep_search=False) # Quick search without deep scan """ return await tools.quick_search( ctx, query=query, search_types=search_types, deep_search=deep_search, page=page, per_page=per_page, )
• productive_client.py:141-156 (helper)

  Supporting API client method in ProductiveClient that constructs query parameters and performs HTTP GET request to Productive's /search/quick endpoint.
  async def quick_search(self, query: str, search_types: Optional[list] = None, deep_search: bool = True, page: int = 1, per_page: int = 50) -> Dict[str, Any]: """Quick search across projects, tasks, pages, and actions""" if search_types is None: search_types = ["action", "project", "task", "page"] params = { "filter[query]": query, "filter[type]": ",".join(search_types), "filter[status]": "all", "filter[deep_search]": str(deep_search).lower(), "page": page, "per_page": per_page } return await self._request("GET", "/search/quick", params=params)