USPTO Final Petition Decisions MCP Server

Instructions

Implementation Reference

• src/fpd_mcp/main.py:689-785 (handler)

  Primary MCP tool handler: decorated with @mcp.tool(name="Search_petitions_by_art_unit"), performs validation, delegates to API client, applies balanced field filtering, adds specialized LLM guidance for art unit analysis.

  @mcp.tool(name="Search_petitions_by_art_unit")
  @async_tool_error_handler("art_unit_search")
  async def fpd_search_petitions_by_art_unit(
      art_unit: str,
      date_range: Optional[str] = None,
      limit: int = 50
  ) -> Dict[str, Any]:
      """Search petitions by art unit number for examiner/art unit quality analysis.
  
  **Use for:** Art unit quality assessment, systematic petition patterns, examiner behavior analysis.
  Returns balanced field set for cross-referencing with PFW examiner data and PTAB challenge rates.
  
  **Example:**
  - fpd_search_petitions_by_art_unit(art_unit="2128", limit=50)
  - fpd_search_petitions_by_art_unit(art_unit="2128", date_range="2020-01-01:2024-12-31")
  
  **Analysis patterns:**
  - High petition frequency → Difficult examiners or challenging technology
  - Frequent revival petitions (37 CFR 1.137) → Docketing/procedural issues
  - Examiner disputes (37 CFR 1.181) → Communication/quality problems
  - Denied petitions → Weak prosecution practices
  
  **Cross-MCP integration:**
  - applicationNumberText → pfw_search_applications_minimal with fields parameter for examiner names
  - Group petitions by examiner to identify individual patterns
  - patentNumber → PTAB MCP to correlate petition history with challenge success
  
  **Parameters:**
  - art_unit: Art unit number (e.g., "2128", "3600")
  - date_range: Optional date range (format: "YYYY-MM-DD:YYYY-MM-DD")
  - limit: Maximum results (default 50, max 200)"""
      try:
          # Input validation
          if not art_unit or len(art_unit.strip()) == 0:
              return format_error_response("Art unit cannot be empty", 400)
          if limit < api_constants.MIN_SEARCH_LIMIT or limit > api_constants.MAX_SEARCH_LIMIT:
              return format_error_response(f"Limit must be between {api_constants.MIN_SEARCH_LIMIT} and {api_constants.MAX_SEARCH_LIMIT}", 400)
          if date_range:
              # Basic date range format validation
              parts = date_range.split(":")
              if len(parts) != 2:
                  return format_error_response(
                      "Date range must be in format YYYY-MM-DD:YYYY-MM-DD", 400
                  )
  
          # Use API client's search_by_art_unit method
          result = await api_client.search_by_art_unit(
              art_unit=art_unit,
              date_range=date_range,
              limit=limit
          )
  
          # Check for errors
          if "error" in result:
              return result
  
          # Filter response using balanced field set
          fields = field_manager.get_fields("petitions_balanced")
          filtered_result = field_manager.filter_response(result, "petitions_balanced")
  
          # Add art unit analysis guidance
          filtered_result["llm_guidance"] = {
              "workflow": "Art Unit Discovery -> Examiner Mapping -> PTAB Correlation",
              "analysis_patterns": {
                  "high_frequency": "Many petitions → Difficult examiners/technology/systematic issues",
                  "revival_clustering": "Multiple 37 CFR 1.137 → Docketing/procedural problems",
                  "examiner_disputes": "Multiple 37 CFR 1.181 → Communication/quality issues",
                  "ptab_correlation": "High petitions + high PTAB invalidation → Quality issues"
              },
              "next_steps": [
                  "Use pfw_search_applications_minimal with fields parameter for examiner mapping",
                  "Group petitions by examiner to identify individual patterns",
                  "Check GRANTED/DENIED outcomes to assess Director overturn rates",
                  "Cross-reference patentNumbers with PTAB for challenge correlation"
              ],
              "red_flags": {
                  "high_denial_rate": "Weak prosecution practices",
                  "multiple_examiners": "Art unit-wide problem",
                  "temporal_clustering": "Process breakdown in specific periods"
              }
          }
  
          return filtered_result
  
      except ValueError as e:
          logger.warning(f"Validation error in art unit search: {str(e)}")
          return format_error_response(str(e), 400)
      except httpx.HTTPStatusError as e:
          logger.error(f"API error in art unit search: {e.response.status_code} - {e.response.text}")
          return format_error_response(f"API error: {e.response.text}", e.response.status_code)
      except httpx.TimeoutException as e:
          logger.error(f"API timeout in art unit search: {str(e)}")
          return format_error_response("Request timeout - please try again", 408)
      except Exception as e:
          logger.error(f"Unexpected error in art unit search: {str(e)}")
          return format_error_response(f"Internal error: {str(e)}", 500)

• src/fpd_mcp/services/fpd_service.py:144-172 (handler)

  Service layer handler that delegates to API client and applies field filtering.

  async def search_by_art_unit(
      self,
      art_unit: str,
      date_range: Optional[str] = None,
      limit: int = 50
  ) -> Dict[str, Any]:
      """
      Search petitions by art unit
  
      Args:
          art_unit: Art unit number
          date_range: Optional date range filter
          limit: Number of results to return
  
      Returns:
          Search results for the art unit
      """
      result = await self.api_client.search_by_art_unit(
          art_unit=art_unit,
          date_range=date_range,
          limit=limit
      )
  
      # Filter response using balanced field set
      if "error" not in result:
          result = self.field_manager.filter_response(result, "petitions_balanced")
  
      return result

• src/fpd_mcp/api/fpd_client.py:339-381 (handler)

  Core API client implementation: constructs Lucene query for art unit search, applies date range filters, delegates to generic search_petitions method.

  async def search_by_art_unit(
      self,
      art_unit: str,
      date_range: Optional[str] = None,
      limit: int = 50
  ) -> Dict[str, Any]:
      """
      Search petitions by art unit number
  
      Args:
          art_unit: Art unit number (e.g., "2128")
          date_range: Optional date range filter (e.g., "2020-01-01:2024-12-31")
          limit: Maximum number of results
  
      Returns:
          Dict containing search results
      """
      try:
          # Build query
          query = f"{QueryFieldNames.ART_UNIT}:{art_unit}"
  
          # Build filters for date range if provided
          filters = []
          if date_range:
              # Parse date range
              parts = date_range.split(":")
              if len(parts) == 2:
                  filters.append({
                      "field": FPDFields.PETITION_MAIL_DATE,
                      "valueFrom": parts[0],
                      "valueTo": parts[1]
                  })
  
          return await self.search_petitions(
              query=query,
              filters=filters if filters else None,
              limit=limit
          )
  
      except Exception as e:
          logger.error(f"Error in search_by_art_unit: {str(e)}")
          return format_error_response(str(e), 500, generate_request_id())

• src/fpd_mcp/main.py:689-689 (registration)

  MCP tool registration decorator specifying the tool name.

  @mcp.tool(name="Search_petitions_by_art_unit")

• src/fpd_mcp/main.py:691-695 (schema)

  Input schema defined by function parameters: art_unit (required str), date_range (optional str), limit (optional int=50). Output: Dict[str, Any] with filtered results and guidance.

  async def fpd_search_petitions_by_art_unit(
      art_unit: str,
      date_range: Optional[str] = None,
      limit: int = 50
  ) -> Dict[str, Any]: