search_packages

Instructions

[COST: 0.02 USDC (Anti-spam fee)]
The primary marketplace exploration tool for USPTO structured patent data.
Currently focuses on IPC Sections G (Physics, e.g., G01) and H (Electricity, e.g., H04).
- If search_query is empty: Returns the full inventory (tags, titles, tech_stacks, prices) sorted by tag.
- If search_query is provided: Returns detailed JSON info including 'description' and 'patent_ids' for deduplication.

Implementation Reference

• mcp_server.py:98-140 (handler)

  Main handler for the search_packages tool. It queries Supabase views (v_package_marketplace or v_catalogs) based on the search query, returning full inventory or detailed search results.

  @mcp.tool()
  def search_packages(search_query: str = Field(default="", description="Search query. Use keywords like 'USPTO', 'G01', 'Physics', or 'Electricity'. Leave blank for all packages.")) -> str:
      """
      [COST: 0.02 USDC (Anti-spam fee)]
      The primary marketplace exploration tool for USPTO structured patent data.
      Currently focuses on IPC Sections G (Physics, e.g., G01) and H (Electricity, e.g., H04).
      - If search_query is empty: Returns the full inventory (tags, titles, tech_stacks, prices) sorted by tag.
      - If search_query is provided: Returns detailed JSON info including 'description' and 'patent_ids' for deduplication.
      """
      if not supabase:
          return json.dumps({"error": "Database connection failed"})
  
      try:
          if search_query in ["null", "None", ""]:
              search_query = ""
              
          search_query = search_query.strip()
  
          if not search_query:
              res = supabase.table("v_package_marketplace") \
                  .select("package_tag, category, title, record_count, price_usd, avg_importance_p, tech_stacks") \
                  .order("package_tag") \
                  .execute()
              return json.dumps(clean_supabase_data(res.data), ensure_ascii=False)
          else:
              safe_query = search_query.replace(",", " ")
              
              # 1. タグ完全一致の高速検索 (詳細データを含む v_catalogs を使用)
              if "_" in safe_query:
                  exact_res = supabase.table("v_catalogs").select("*").eq("package_tag", safe_query).execute()
                  if exact_res.data:
                      return json.dumps(clean_supabase_data(exact_res.data), ensure_ascii=False)
  
              # 2. 通常のあいまい検索 (LIMITを設けてタイムアウト防止)
              res = supabase.table("v_catalogs") \
                  .select("*") \
                  .or_(f"category.ilike.%{safe_query}%,title.ilike.%{safe_query}%,description.ilike.%{safe_query}%,package_tag.ilike.%{safe_query}%") \
                  .order("package_tag") \
                  .limit(20) \
                  .execute()
              return json.dumps(clean_supabase_data(res.data), ensure_ascii=False)
      except Exception as e:
          return json.dumps({"error": str(e)})

• mcp_server.py:98-98 (registration)

  Registration: The @mcp.tool() decorator registers search_packages as an MCP tool on the FastMCP instance.

  @mcp.tool()

• mcp_server.py:84-93 (helper)

  Helper function that parses string-encoded JSON arrays from Supabase into native Python lists before returning results.

  def clean_supabase_data(rows):
      array_fields = ['assignee', 'inventor', 'secondary_cpcs', 'attr_tech_stack', 'tech_stacks', 'biz_target_ind', 'attr_performance', 'package_tags', 'cited_patents']
      for row in rows:
          for field in array_fields:
              if field in row and isinstance(row[field], str):
                  try:
                      row[field] = json.loads(row[field])
                  except Exception:
                      pass
      return rows