USPTO Final Petition Decisions MCP Server

Instructions

Implementation Reference

• src/fpd_mcp/main.py:992-1326 (handler)

  The primary handler function implementing the FPD_get_document_download tool. It validates inputs, detects centralized proxy (PFW MCP), generates persistent or session-based download URLs, retrieves document metadata from USPTO API, registers with proxy if applicable, and provides LLM-formatted response with clickable links and guidance.

  @mcp.tool(name="FPD_get_document_download")
  @async_tool_error_handler("document_download")
  async def fpd_get_document_download(
      petition_id: str,
      document_identifier: str,
      proxy_port: Optional[int] = None,
      generate_persistent_link: bool = True
  ) -> Dict[str, Any]:
      """Generate browser-accessible download URL for petition documents (PDFs) via secure proxy.
  
  **ALWAYS-ON PROXY (DEFAULT):** Proxy server starts with MCP - download links work immediately.
  
  **Workflow:**
  1. fpd_get_petition_details(petition_id='uuid', include_documents=True) → get documentBag
  2. fpd_get_document_download(petition_id='uuid', document_identifier='ABC123') → get download link
  3. Provide download link to user
  
  **CRITICAL RESPONSE FORMAT - Always format with BOTH clickable link and raw URL:**
  **📁 [Download {DocumentType} ({PageCount} pages)]({proxy_url})** | Raw URL: `{proxy_url}`
  
  Why both formats?
  - Clickable links work in Claude Desktop and most clients
  - Raw URLs enable copy/paste in Msty and other clients where links aren't clickable
  
  **Document types:**
  - Petition document: Original petition filed with USPTO
  - Decision document: Director's final decision
  - Supporting exhibits: Declarations, prior art, technical documents
  
  **Parameters:**
  - petition_id: Petition UUID from search results
  - document_identifier: Document identifier from documentBag
  - proxy_port: Optional (defaults to FPD_PROXY_PORT env var or 8081)
  - generate_persistent_link: Generate 7-day persistent link (default: True)
    - True: Attempts persistent link via USPTO PFW MCP (works across MCP restarts)
    - False: Session-based link (works while MCP running, no PFW required)"""
      try:
          # Input validation
          if not petition_id or len(petition_id.strip()) == 0:
              return format_error_response("Petition ID cannot be empty", 400)
          if not document_identifier or len(document_identifier.strip()) == 0:
              return format_error_response("Document identifier cannot be empty", 400)
  
          # Handle persistent link generation (requires PFW MCP)
          if generate_persistent_link:
              centralized_port = os.getenv('CENTRALIZED_PROXY_PORT', '').lower()
  
              if centralized_port and centralized_port != 'none':
                  # PFW centralized proxy detected - forward to PFW for persistent link
                  try:
                      pfw_port = int(centralized_port)
                      logger.info(f"Generating persistent link via centralized USPTO PFW proxy (port {pfw_port})")
  
                      # Construct persistent link request to PFW proxy
                      # PFW proxy should have an endpoint for generating persistent links
                      # Format: POST http://localhost:8080/persistent-link
                      persistent_link_url = f"http://localhost:{pfw_port}/persistent-link"
  
                      async with httpx.AsyncClient() as client:
                          response = await client.post(
                              persistent_link_url,
                              json={
                                  "source": "fpd",
                                  "petition_id": petition_id,
                                  "document_identifier": document_identifier,
                                  "expires_days": 7
                              },
                              timeout=30.0
                          )
  
                          if response.status_code == 200:
                              result = response.json()
                              return {
                                  "success": True,
                                  "persistent_download_url": result.get("persistent_url"),
                                  "expires_in_days": 7,
                                  "note": "Generated via centralized USPTO PFW proxy - works across MCP restarts",
                                  "ecosystem_integration": "Using PFW centralized database for persistent links"
                              }
                          else:
                              # PFW proxy doesn't support persistent links yet
                              logger.warning(f"PFW proxy persistent link generation failed: {response.status_code}")
                              # Fall through to immediate link with note
  
                  except Exception as e:
                      logger.warning(f"Failed to generate persistent link via PFW: {e}")
                      # Fall through to immediate link with note
  
              # No centralized proxy or persistent link generation failed
              # Return helpful message encouraging PFW installation
              if not centralized_port:
                  return {
                      "success": False,
                      "error": "Persistent links require USPTO PFW MCP for centralized database",
                      "suggestion": "Install USPTO PFW MCP for persistent links, or use immediate links (generate_persistent_link=false)",
                      "immediate_alternative": f"Call this tool with generate_persistent_link=false for session-based download link",
                      "pfw_benefits": [
                          "7-day persistent encrypted links (work across MCP restarts)",
                          "Centralized proxy server (unified rate limiting)",
                          "Cross-MCP document sharing and caching",
                          "Complete USPTO prosecution + petition workflow"
                      ],
                      "note": "FPD provides immediate downloads only - PFW provides persistent links + enhanced features",
                      "recommendation": "Install both USPTO FPD + PFW MCPs for complete patent lifecycle analysis"
                  }
              else:
                  # PFW is available but persistent link generation failed - fallback to immediate link
                  logger.info("Persistent link generation not available, falling back to immediate link")
  
          # Enhanced proxy port detection with centralized proxy support
          if proxy_port is None:
              # Check if centralized proxy is available (and not "none")
              centralized_port = os.getenv('CENTRALIZED_PROXY_PORT', '').lower()
              if centralized_port and centralized_port != 'none':
                  proxy_port = int(centralized_port)
                  logger.info(f"Using centralized USPTO proxy on port {proxy_port}")
              else:
                  # Check FPD_PROXY_PORT first (MCP-specific), then PROXY_PORT (generic)
                  proxy_port = get_local_proxy_port()
                  logger.info(f"Using local FPD proxy on port {proxy_port}")
  
          # Start proxy server if not already running (unless using centralized proxy)
          centralized_port_check = os.getenv('CENTRALIZED_PROXY_PORT', '').lower()
          if not centralized_port_check or centralized_port_check == 'none':
              await _ensure_proxy_server_running(proxy_port)
          else:
              # Centralized proxy is already running (managed by PFW MCP)
              logger.info("Using centralized proxy - no local proxy startup needed")
  
          # Construct proxy URL (port 8081 to avoid conflict with PFW proxy on 8080)
          proxy_url = f"http://localhost:{proxy_port}/download/{petition_id}/{document_identifier}"
  
          # Also construct direct API URL for reference
          direct_url = f"{api_client.base_url}/{petition_id}/documents/{document_identifier}"
  
          # Get petition details to find document metadata
          petition_result = await api_client.get_petition_by_id(petition_id, include_documents=True)
  
          if "error" in petition_result:
              return petition_result
  
          # Extract from nested structure
          petition_data = petition_result.get(FPDFields.PETITION_DECISION_DATA_BAG, [])
          if not petition_data:
              return format_error_response("Petition data not found", 404)
  
          # Get documentBag from first petition in array
          documents = petition_data[0].get(FPDFields.DOCUMENT_BAG, [])
  
          # Find document metadata
          document_metadata = None
          for doc in documents:
              if doc.get(FPDFields.DOCUMENT_IDENTIFIER) == document_identifier:
                  document_metadata = doc
                  break
  
          if not document_metadata:
              return format_error_response(
                  f"Document {document_identifier} not found in petition {petition_id}", 404
              )
  
          # Track if centralized proxy registration succeeds
          centralized_registration_success = False
  
          # Register document with centralized proxy if using PFW
          # Check if CENTRALIZED_PROXY_PORT is set and not "none"
          centralized_port_env = os.getenv('CENTRALIZED_PROXY_PORT', '').lower()
          if centralized_port_env and centralized_port_env != 'none':
              # Extract PDF download URL from document metadata
              download_options = document_metadata.get(FPDFields.DOWNLOAD_OPTION_BAG, [])
              pdf_download_url = None
  
              for option in download_options:
                  if option.get(FPDFields.MIME_TYPE_IDENTIFIER) == 'PDF':
                      pdf_download_url = option.get(FPDFields.DOWNLOAD_URL)
                      break
  
              if pdf_download_url:
                  # Extract metadata for enhanced filename generation
                  petition_mail_date = petition_data[0].get(FPDFields.PETITION_MAIL_DATE)
                  app_number = petition_data[0].get(FPDFields.APPLICATION_NUMBER_TEXT)
                  patent_number = petition_data[0].get(FPDFields.PATENT_NUMBER)
                  doc_description = document_metadata.get(FPDFields.DOCUMENT_CODE_DESCRIPTION_TEXT)
                  doc_code = document_metadata.get(FPDFields.DOCUMENT_CODE)
  
                  # Generate enhanced filename using local proxy logic
                  enhanced_filename = generate_enhanced_filename(
                      petition_mail_date=petition_mail_date,
                      app_number=app_number,
                      patent_number=patent_number,
                      document_description=doc_description,
                      document_code=doc_code,
                      max_desc_length=40
                  )
  
                  logger.info(f"Generated enhanced filename for PFW registration: {enhanced_filename}")
  
                  try:
                      # Register FPD document with PFW centralized proxy
                      # Use the already validated centralized_port_env variable
                      pfw_port = int(centralized_port_env)
                      register_url = f"http://localhost:{pfw_port}/register-fpd-document"
  
                      # Create secure token for document access
                      access_token = mcp_auth.create_document_access_token(
                          petition_id=petition_id,
                          document_identifier=document_identifier,
                          application_number=app_number
                      )
  
                      async with httpx.AsyncClient() as client:
                          response_reg = await client.post(
                              register_url,
                              json={
                                  "source": "fpd",
                                  "petition_id": petition_id,
                                  "document_identifier": document_identifier,
                                  "download_url": pdf_download_url,
                                  "access_token": access_token,  # Secure token instead of raw API key
                                  "application_number": app_number,
                                  "enhanced_filename": enhanced_filename  # Professional filename for downloads
                              },
                              timeout=5.0
                          )
  
                          if response_reg.status_code == 200:
                              logger.info(f"✅ Successfully registered FPD document with centralized proxy")
                              centralized_registration_success = True
                          else:
                              logger.warning(
                                  f"❌ Failed to register document with centralized proxy: HTTP {response_reg.status_code}"
                              )
                              try:
                                  error_detail = response_reg.json()
                                  logger.warning(f"   Registration error details: {error_detail}")
                              except Exception:
                                  logger.warning(f"   Response body: {response_reg.text[:500]}")
  
                  except Exception as e:
                      logger.warning(f"❌ Failed to register document with centralized proxy: {e}")
  
          # Implement fallback: if centralized registration failed, use local proxy
          # Only applies if we actually tried to use centralized proxy (not "none")
          centralized_port_check = os.getenv('CENTRALIZED_PROXY_PORT', '').lower()
          if centralized_port_check and centralized_port_check != 'none' and not centralized_registration_success:
              logger.warning("⚠️  Centralized proxy registration failed - falling back to local FPD proxy")
              # Start local proxy as fallback
              local_proxy_port = get_local_proxy_port()
              await _ensure_proxy_server_running(local_proxy_port)
              # Update proxy URL to use local proxy
              proxy_url = f"http://localhost:{local_proxy_port}/download/{petition_id}/{document_identifier}"
              logger.info(f"🔄 Using local FPD proxy on port {local_proxy_port} for this download")
  
          # Determine proxy type for response metadata
          centralized_port_check = os.getenv('CENTRALIZED_PROXY_PORT', '').lower()
          proxy_type = "centralized" if (centralized_port_check and centralized_port_check != 'none' and centralized_registration_success) else "local"
          proxy_port_used = proxy_port if proxy_type == "centralized" else get_local_proxy_port()
  
          # Build response with LLM guidance for clickable links
          response = {
              "success": True,
              "petition_id": petition_id,
              "document_identifier": document_identifier,
              "proxy_download_url": proxy_url,
              "direct_url": direct_url,
              "document_info": document_metadata,
  
              # Proxy information for debugging
              "proxy_info": {
                  "type": proxy_type,
                  "port": proxy_port_used,
                  "status": "centralized_registered" if centralized_registration_success else "local_fallback"
              },
  
              # NEW: Explicit LLM guidance for proper response formatting
              "llm_response_guidance": {
                  "critical_requirement": "ALWAYS provide BOTH clickable markdown link AND raw URL",
                  "required_format": f"**📁 [Download {document_metadata.get('documentFileName', 'Document')} ({document_metadata.get('pageCount', 'N/A')} pages)]({proxy_url})** | Raw URL: `{proxy_url}`",
                  "user_expectation": "User requested a download - they need immediate browser access to the PDF",
                  "wrong_response": "Don't just show the raw URL or tool results",
                  "correct_response": "Format as clickable markdown link with document description and page count PLUS raw URL for copy/paste",
                  "explanation": "Clickable link works in Claude Desktop, raw URL enables copy/paste in Msty and other clients where links aren't clickable"
              },
  
              "access_instructions": {
                  "method": "Proxy server download (recommended) or direct API access",
                  "proxy_url": f"{proxy_url} - Click to download via secure proxy",
                  "proxy_port": proxy_port_used,
                  "proxy_note": f"Proxy handles USPTO API authentication ({proxy_type} proxy on port {proxy_port_used})",
                  "rate_limit": "USPTO allows 5 downloads per 10 seconds",
                  "file_type": "PDF document",
                  "estimated_size": f"{document_metadata.get('pageCount', 'unknown')} pages"
              },
  
              "llm_guidance": {
                  "next_steps": [
                      f"Present proxy download URL to user: ** [Download {document_metadata.get('documentFileName', 'Document')} ({document_metadata.get('pageCount', 'unknown')} pages)]({proxy_url})**",
                      "Proxy server is now running on port 8081 (started automatically)",
                      "User can click the link to download PDF directly through secure proxy",
                      "Proxy server handles USPTO API authentication automatically"
                  ],
                  "document_context": {
                      "petition_type": petition_data[0].get(FPDFields.DECISION_PETITION_TYPE_CODE_DESCRIPTION_TEXT, "Unknown"),
                      "decision_outcome": petition_data[0].get(FPDFields.DECISION_TYPE_CODE_DESCRIPTION_TEXT, "Unknown"),
                      "decision_date": petition_data[0].get(FPDFields.DECISION_DATE, "Unknown")
                  }
              },
  
              # Critical UX reminder
              "ux_critical": "The user wants this PDF file - make the download link immediately clickable!",
  
              # Response validation hints
              "response_validation": {
                  "check_for_markdown_link": "Response should contain [text](url) format",
                  "check_for_clickable_emoji": "Should start with  emoji for visual recognition",
                  "check_for_description": "Link text should describe the document type and page count",
                  "success_pattern": f"** [Download {document_metadata.get('documentFileName', 'DocumentType')} ({document_metadata.get('pageCount', 'N')} pages)](http://localhost:{proxy_port}/download/...)**"
              }
          }
  
          return response
  
      except ValueError as e:
          logger.warning(f"Validation error in get document download: {str(e)}")
          return format_error_response(str(e), 400)
      except httpx.HTTPStatusError as e:
          logger.error(f"API error in get document download: {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 get document download: {str(e)}")
          return format_error_response("Request timeout - please try again", 408)
      except Exception as e:
          logger.error(f"Unexpected error in get document download: {str(e)}")
          return format_error_response(f"Internal error: {str(e)}", 500)

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

  MCP tool registration decorator for FPD_get_document_download.

  @mcp.tool(name="FPD_get_document_download")

• src/fpd_mcp/main.py:1537-1565 (helper)

  Helper function to run the proxy server (FastAPI app) which serves the /download/{petition_id}/{document_identifier} endpoint used by the tool's generated URLs.

  async def _run_proxy_server(port: int = 8081):
      """Run the FastAPI proxy server
  
      Uses API key from Settings (which may come from secure storage or environment variables)
      """
      try:
          import uvicorn
          from .proxy.server import create_proxy_app
  
          # Pass API key and port from Settings to proxy server
          # This allows proxy to work with secure storage (Windows DPAPI)
          app = create_proxy_app(api_key=settings.uspto_api_key, port=port)
          config = uvicorn.Config(
              app,
              host="127.0.0.1",
              port=port,
              log_level="info",
              access_log=False  # Reduce noise in logs
          )
          server = uvicorn.Server(config)
          logger.info(f"HTTP proxy server starting on http://127.0.0.1:{port}")
          await server.serve()
  
      except Exception as e:
          global _proxy_server_running
          _proxy_server_running = False
          logger.error(f"Proxy server failed: {e}")
          raise

• src/fpd_mcp/main.py:1512-1536 (helper)

  Helper function that ensures the proxy server is started automatically when the tool is called (always-on or on-demand mode).

  async def _ensure_proxy_server_running(port: int = 8081):
      """Ensure the proxy server is running (auto-start on first download)"""
      global _proxy_server_running, _proxy_server_task
  
      if not _proxy_server_running:
          logger.info(f"Starting HTTP proxy server on port {port}")
  
          # Wrap background task with exception handler
          async def safe_proxy_runner():
              try:
                  await _run_proxy_server(port)
              except Exception as e:
                  logger.error(f"Proxy server crashed: {e}", exc_info=True)
                  global _proxy_server_running
                  _proxy_server_running = False
                  # Allow graceful degradation - main server continues without proxy
  
          _proxy_server_task = asyncio.create_task(safe_proxy_runner())
          _proxy_server_running = True
          # Give the server a moment to start
          await asyncio.sleep(0.5)
          logger.info(f"Proxy server started successfully on port {port}")