""" Genesis Patent MCP Server ========================== FastMCP server exposing patent research, drafting, portfolio tracking, and IP strategy tools to Claude. Port: 8901 (voice-bridge uses 8900, patent-os uses stdio) Auth: Bearer token via GENESIS_PATENT_MCP_TOKEN env var Transport: SSE (default) or HTTP Tools: search_patents - Search USPTO / Google Patents analyze_patent_claims - Break down and scope patent claims draft_patent_claims - Draft independent + dependent claims check_prior_art - Find prior art for an invention track_portfolio - Read/write patent portfolio tracker generate_ip_memo - Produce IP strategy memos calculate_fees - USPTO fee schedules monitor_competitor_patents - Recent filings by competitor """ import os import sys import json import logging import re import time import hashlib from datetime import datetime, timedelta from pathlib import Path from typing import Optional import requests from fastmcp import FastMCP from starlette.middleware import Middleware from starlette.requests import Request from starlette.responses import JSONResponse from starlette.types import ASGIApp, Receive, Scope, Send # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(name)s] %(levelname)s %(message)s", handlers=[logging.StreamHandler()], ) logger = logging.getLogger("genesis.patent_mcp") # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- AUTH_TOKEN = os.environ.get( "GENESIS_PATENT_MCP_TOKEN", "genesis-patent-mcp-2026-dev-key", ) HOST = os.environ.get("GENESIS_PATENT_MCP_HOST", "0.0.0.0") PORT = int(os.environ.get("GENESIS_PATENT_MCP_PORT", "8901")) TRANSPORT = os.environ.get("GENESIS_PATENT_MCP_TRANSPORT", "sse") # Paths — all on E: drive, NO SQLite, NO C: drive GENESIS_ROOT = Path(os.environ.get("GENESIS_ROOT", "E:/genesis-system")) PORTFOLIO_FILE = Path(os.environ.get( "GENESIS_PATENT_PORTFOLIO_FILE", str(GENESIS_ROOT / "data" / "patent_portfolio.json"), )) KG_ENTITIES_DIR = GENESIS_ROOT / "KNOWLEDGE_GRAPH" / "entities" # USPTO Open Data API (no key required for basic search) USPTO_SEARCH_URL = "https://developer.uspto.gov/ibd-api/v1/application/publications" USPTO_PATENT_URL = "https://api.patentsview.org/patents/query" # Google Patents Scraping (fallback) GOOGLE_PATENTS_BASE = "https://patents.google.com" # --------------------------------------------------------------------------- # Portfolio helpers — JSON file, no SQLite # --------------------------------------------------------------------------- def _load_portfolio() -> dict: """Load patent portfolio from JSON file.""" if not PORTFOLIO_FILE.exists(): return {"patents": [], "last_updated": None, "version": "1.0.0"} try: with open(PORTFOLIO_FILE, "r", encoding="utf-8") as f: return json.load(f) except Exception as e: logger.error("Failed to load portfolio: %s", e) return {"patents": [], "last_updated": None, "version": "1.0.0"} def _save_portfolio(portfolio: dict) -> bool: """Save patent portfolio to JSON file.""" try: PORTFOLIO_FILE.parent.mkdir(parents=True, exist_ok=True) portfolio["last_updated"] = datetime.utcnow().isoformat() with open(PORTFOLIO_FILE, "w", encoding="utf-8") as f: json.dump(portfolio, f, indent=2) return True except Exception as e: logger.error("Failed to save portfolio: %s", e) return False # --------------------------------------------------------------------------- # Bearer Token Auth Middleware # --------------------------------------------------------------------------- class BearerTokenMiddleware: """ASGI middleware enforcing Bearer auth on /mcp and /sse paths.""" OPEN_PATHS = {"/health", "/favicon.ico"} def __init__(self, app: ASGIApp, token: str): self.app = app self.token = token async def __call__(self, scope: Scope, receive: Receive, send: Send): if scope["type"] == "http": path = scope.get("path", "") if path in self.OPEN_PATHS: await self.app(scope, receive, send) return headers = dict(scope.get("headers", [])) auth_header = headers.get(b"authorization", b"").decode("utf-8", errors="ignore") if not auth_header.startswith("Bearer "): response = JSONResponse( {"error": "Missing or invalid Authorization header"}, status_code=401, ) await response(scope, receive, send) return token = auth_header[7:] if token != self.token: response = JSONResponse( {"error": "Invalid bearer token"}, status_code=403, ) await response(scope, receive, send) return if path == "/sse": async def send_no_buffer(message): if message["type"] == "http.response.start": headers_list = list(message.get("headers", [])) headers_list.append((b"x-accel-buffering", b"no")) headers_list.append((b"cache-control", b"no-cache, no-transform")) message = {**message, "headers": headers_list} await send(message) await self.app(scope, receive, send_no_buffer) return await self.app(scope, receive, send) # --------------------------------------------------------------------------- # FastMCP Server # --------------------------------------------------------------------------- mcp = FastMCP( "Genesis Patent MCP", instructions=( "Genesis Patent MCP provides tools for patent research, claim drafting, " "prior art searches, portfolio tracking, IP strategy memos, USPTO fee " "calculation, and competitor patent monitoring. Use these tools to help " "inventors protect their innovations, track their IP portfolio, and make " "strategic patenting decisions." ), ) # --------------------------------------------------------------------------- # Health Check # --------------------------------------------------------------------------- @mcp.custom_route("/health", methods=["GET"]) async def health_check(request: Request): """Health check endpoint — no auth required.""" portfolio = _load_portfolio() patent_count = len(portfolio.get("patents", [])) return JSONResponse({ "status": "healthy", "service": "genesis-patent-mcp", "version": "1.0.0", "port": PORT, "patent_count": patent_count, "timestamp": datetime.utcnow().isoformat(), }) # --------------------------------------------------------------------------- # Tool 1: search_patents # --------------------------------------------------------------------------- @mcp.tool def search_patents( query: str, date_from: str = "", date_to: str = "", assignee: str = "", limit: int = 10, ) -> str: """ Search patent databases (USPTO PatentsView API + Google Patents) for relevant patents. Args: query: Search query string — keywords describing the invention or technology. date_from: Start date filter in YYYY-MM-DD format (optional). date_to: End date filter in YYYY-MM-DD format (optional). assignee: Filter by patent assignee/company name (optional). limit: Maximum number of results to return (default 10, max 25). Returns: List of matching patents with numbers, titles, abstracts, and links. """ limit = min(max(1, limit), 25) results = [] # Try USPTO PatentsView API (free, no key required) try: query_params = { "q": {}, "f": ["patent_number", "patent_title", "patent_abstract", "patent_date", "assignee_organization", "inventor_last_name"], "o": {"patent_date": "desc"}, "s": [{"patent_date": "desc"}], } # Build query filter conditions = [] if query: conditions.append({"_text_phrase": {"patent_abstract": query}}) if assignee: conditions.append({"_contains": {"assignee_organization": assignee}}) if date_from: conditions.append({"_gte": {"patent_date": date_from}}) if date_to: conditions.append({"_lte": {"patent_date": date_to}}) if conditions: query_params["q"] = {"_and": conditions} if len(conditions) > 1 else conditions[0] else: query_params["q"] = {"_text_phrase": {"patent_abstract": query}} response = requests.post( USPTO_PATENT_URL, json=query_params, headers={"Content-Type": "application/json"}, timeout=15, ) if response.status_code == 200: data = response.json() patents_data = data.get("patents") or [] for p in patents_data[:limit]: abstract = p.get("patent_abstract", "") or "" results.append({ "patent_number": p.get("patent_number", ""), "title": p.get("patent_title", ""), "date": p.get("patent_date", ""), "assignee": (p.get("assignees") or [{}])[0].get("assignee_organization", ""), "abstract": abstract[:400] + "..." if len(abstract) > 400 else abstract, "link": f"https://patents.google.com/patent/US{p.get('patent_number', '')}", "source": "USPTO PatentsView", }) else: logger.warning("USPTO PatentsView API returned %s", response.status_code) except Exception as e: logger.warning("USPTO PatentsView search failed: %s", e) # Fallback: USPTO full-text search via developer API if not results: try: params = { "searchText": query, "start": 0, "rows": limit, } if date_from: params["dateRangeData.startDate"] = date_from if date_to: params["dateRangeData.endDate"] = date_to response = requests.get( "https://developer.uspto.gov/ibd-api/v1/application/publications", params=params, headers={"Accept": "application/json"}, timeout=15, ) if response.status_code == 200: data = response.json() docs = data.get("results", {}).get("publicationSearchResult", {}).get("docs", []) for doc in docs[:limit]: abstract = doc.get("abstractText", "") if isinstance(abstract, list): abstract = " ".join(str(a) for a in abstract) results.append({ "patent_number": doc.get("patentNumber", doc.get("applicationNumberText", "")), "title": doc.get("inventionTitle", ""), "date": doc.get("publicationDate", doc.get("filingDate", "")), "assignee": doc.get("assigneeNameText", ""), "abstract": str(abstract)[:400] + "..." if len(str(abstract)) > 400 else str(abstract), "link": f"https://patents.google.com/?q={doc.get('patentNumber', '')}", "source": "USPTO Publications", }) except Exception as e: logger.warning("USPTO Publications search failed: %s", e) if not results: return ( f"No patents found for query: '{query}'. " f"Try broader terms or check USPTO directly at https://patents.google.com/?q={requests.utils.quote(query)}" ) output = f"Patent Search Results for: '{query}'\n" output += f"Found {len(results)} patent(s)\n" output += "=" * 60 + "\n\n" for i, p in enumerate(results, 1): output += f"{i}. {p['title']}\n" output += f" Number: US{p['patent_number']}\n" output += f" Date: {p['date']}\n" if p.get("assignee"): output += f" Assignee: {p['assignee']}\n" output += f" Abstract: {p['abstract']}\n" output += f" Link: {p['link']}\n" output += f" Source: {p['source']}\n\n" output += f"\nSearch Google Patents directly: https://patents.google.com/?q={requests.utils.quote(query)}" return output # --------------------------------------------------------------------------- # Tool 2: analyze_patent_claims # --------------------------------------------------------------------------- @mcp.tool def analyze_patent_claims( patent_number: str = "", claims_text: str = "", ) -> str: """ Analyze patent claims — identify independent vs dependent claims, scope, and potential weaknesses or strengths. Provide EITHER a patent number (e.g. 'US10123456') OR paste the claims text directly. Args: patent_number: USPTO patent number (e.g. 'US10123456' or '10123456'). claims_text: Raw claims text to analyze (use this if patent_number unavailable). Returns: Structured analysis of claims including independent/dependent breakdown, scope assessment, and strategic notes. """ raw_claims = claims_text.strip() # If patent number provided, try to fetch claims from PatentsView if patent_number and not raw_claims: num = re.sub(r"[^0-9]", "", patent_number) try: response = requests.post( USPTO_PATENT_URL, json={ "q": {"patent_number": num}, "f": ["patent_number", "patent_title", "patent_abstract", "patent_date", "patent_num_claims"], }, headers={"Content-Type": "application/json"}, timeout=15, ) if response.status_code == 200: data = response.json() patents = data.get("patents") or [] if patents: p = patents[0] raw_claims = ( f"[Claims fetched for patent US{num}: " f"{p.get('patent_title', 'Unknown')} " f"(filed {p.get('patent_date', 'unknown date')}, " f"{p.get('patent_num_claims', '?')} claims total)]\n\n" f"Abstract: {p.get('patent_abstract', 'N/A')}" ) except Exception as e: logger.warning("Failed to fetch patent %s: %s", patent_number, e) raw_claims = f"[Could not fetch claims for {patent_number} — analyze manually]" if not raw_claims: return "Error: Provide either a patent_number or claims_text to analyze." # Parse claims structure claim_blocks = re.split(r'\n\s*(?=\d+\.\s)', raw_claims) independent_claims = [] dependent_claims = [] for block in claim_blocks: block = block.strip() if not block: continue # A claim is dependent if it references another claim number is_dependent = bool(re.search( r'\b(claim|claims)\s+\d+', block, re.IGNORECASE )) claim_num_match = re.match(r'^(\d+)\.', block) claim_num = claim_num_match.group(1) if claim_num_match else "?" if is_dependent: dependent_claims.append((claim_num, block[:200])) else: independent_claims.append((claim_num, block[:200])) # Build analysis output = "PATENT CLAIMS ANALYSIS\n" output += "=" * 60 + "\n\n" if patent_number: output += f"Patent: US{re.sub(r'[^0-9]', '', patent_number)}\n" output += f"Google Patents: https://patents.google.com/patent/US{re.sub(r'[^0-9]', '', patent_number)}\n\n" output += f"CLAIM STRUCTURE SUMMARY\n" output += f" Total claim blocks parsed: {len(independent_claims) + len(dependent_claims)}\n" output += f" Independent claims: {len(independent_claims)}\n" output += f" Dependent claims: {len(dependent_claims)}\n\n" if independent_claims: output += "INDEPENDENT CLAIMS (define core scope of protection):\n" for num, text in independent_claims[:5]: output += f"\n Claim {num}: {text[:300]}{'...' if len(text) > 300 else ''}\n" if dependent_claims: output += "\nDEPENDENT CLAIMS (narrowing fallback positions):\n" for num, text in dependent_claims[:8]: output += f"\n Claim {num}: {text[:200]}{'...' if len(text) > 200 else ''}\n" output += "\n\nSCOPE ASSESSMENT:\n" # Assess claim scope heuristics broad_indicators = ["comprising", "including", "wherein", "at least one", "one or more"] narrow_indicators = ["consisting of", "specifically", "exactly", "only", "must"] full_text = raw_claims.lower() broad_count = sum(1 for word in broad_indicators if word in full_text) narrow_count = sum(1 for word in narrow_indicators if word in full_text) if broad_count > narrow_count: scope_verdict = "BROAD — Claims use open-ended language ('comprising', 'including'). Strong protection but higher invalidity risk." elif narrow_count > broad_count: scope_verdict = "NARROW — Claims use closed language ('consisting of'). Easier to validate but easier for competitors to design around." else: scope_verdict = "MODERATE — Mix of open and closed language. Review each independent claim individually." output += f" Scope: {scope_verdict}\n" output += f" Broad language markers found: {broad_count}\n" output += f" Narrow language markers found: {narrow_count}\n" output += "\n\nSTRATEGIC NOTES:\n" if not independent_claims: output += " WARNING: No independent claims detected. Every patent needs at least one independent claim.\n" if len(independent_claims) > 3: output += f" NOTE: {len(independent_claims)} independent claims — more independent claims = more filing fees at USPTO.\n" if dependent_claims: output += f" GOOD: {len(dependent_claims)} dependent claims provide fallback positions if independent claims are invalidated.\n" output += "\n Recommendation: Have a registered patent attorney review claims before filing. " output += "This analysis is informational only, not legal advice.\n" return output # --------------------------------------------------------------------------- # Tool 3: draft_patent_claims # --------------------------------------------------------------------------- @mcp.tool def draft_patent_claims( invention_description: str, invention_type: str = "software", focus_area: str = "", ) -> str: """ Draft patent claims for an invention — produces one independent claim and five dependent claims in proper USPTO format. Args: invention_description: Detailed description of what the invention does, how it works, and what problem it solves. More detail = better claims. invention_type: Type of invention — 'software', 'hardware', 'method', 'system', 'composition', 'device' (default: 'software'). focus_area: Optional specific aspect to emphasize in claims (e.g. 'machine learning', 'data privacy', 'real-time processing'). Returns: Draft independent claim + 5 dependent claims in USPTO patent format. """ if not invention_description.strip(): return "Error: invention_description is required." invention_type = invention_type.lower().strip() # Determine claim preamble preambles = { "software": "A computer-implemented method comprising:", "hardware": "A device comprising:", "method": "A method comprising:", "system": "A system comprising:", "composition": "A composition comprising:", "device": "A device comprising:", } preamble = preambles.get(invention_type, "A computer-implemented method comprising:") # Extract key concepts from description desc = invention_description.strip() words = re.findall(r'\b[a-zA-Z]{4,}\b', desc.lower()) word_freq = {} stop_words = {"that", "this", "with", "from", "have", "been", "will", "when", "where", "which", "their", "them", "they", "using", "used", "each", "into", "through", "about", "some", "also", "then", "can", "data", "system", "method", "process", "based"} for w in words: if w not in stop_words: word_freq[w] = word_freq.get(w, 0) + 1 top_concepts = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)[:8] concept_words = [c[0] for c in top_concepts] # Build claim skeleton focus_clause = f" particularly for {focus_area}" if focus_area else "" output = f"DRAFT PATENT CLAIMS\n" output += f"Invention Type: {invention_type.upper()}\n" if focus_area: output += f"Focus Area: {focus_area}\n" output += "=" * 60 + "\n\n" output += "DISCLAIMER: These are AI-generated draft claims for initial ideation only.\n" output += "They MUST be reviewed and refined by a registered patent attorney before filing.\n" output += "Patent claims require precise legal language — treat these as a starting point.\n\n" output += "=" * 60 + "\n\n" # Independent Claim 1 output += f"1. {preamble}\n\n" if invention_type in ("software", "method"): output += f" receiving, by a processor, input data relating to {concept_words[0] if concept_words else 'a subject matter'}{focus_clause};\n\n" output += f" processing, by the processor, the input data using {concept_words[1] if len(concept_words) > 1 else 'a processing algorithm'} to generate processed output;\n\n" output += f" determining, by the processor, at least one result based on the processed output, wherein the result comprises {concept_words[2] if len(concept_words) > 2 else 'actionable information'};\n\n" output += f" storing, in a non-transitory computer-readable medium, the at least one result; and\n\n" output += f" transmitting, via a network interface, the at least one result to one or more client devices.\n\n" elif invention_type in ("system",): output += f" one or more processors configured to process {concept_words[0] if concept_words else 'data'}{focus_clause};\n\n" output += f" a memory storing instructions that, when executed by the one or more processors, cause the processors to:\n\n" output += f" receive input relating to {concept_words[1] if len(concept_words) > 1 else 'an operation'},\n\n" output += f" apply {concept_words[2] if len(concept_words) > 2 else 'a computation'} to generate at least one output, and\n\n" output += f" transmit the at least one output to a remote system; and\n\n" output += f" a network interface configured to communicate with the remote system.\n\n" else: output += f" a first component configured to receive {concept_words[0] if concept_words else 'input'}{focus_clause};\n\n" output += f" a second component operably coupled to the first component, the second component configured to process the received input; and\n\n" output += f" a third component configured to output a result based on the processed input.\n\n" # Dependent Claims 2-6 output += f"2. The {invention_type} of claim 1, wherein the {concept_words[0] if concept_words else 'input data'} comprises at least one of structured data, unstructured data, or semi-structured data.\n\n" output += f"3. The {invention_type} of claim 1, wherein the processing step further comprises applying one or more {concept_words[1] if len(concept_words) > 1 else 'algorithms'} selected from the group consisting of machine learning, natural language processing, and statistical analysis.\n\n" output += f"4. The {invention_type} of claim 1, wherein the {concept_words[2] if len(concept_words) > 2 else 'result'} is generated in real-time with a latency of less than 500 milliseconds.\n\n" output += f"5. The {invention_type} of claim 1, further comprising encrypting, by the processor, the at least one result prior to transmitting, using an encryption standard selected from AES-256, RSA-4096, or a post-quantum cryptography algorithm.\n\n" output += f"6. The {invention_type} of claim 1, wherein the {concept_words[3] if len(concept_words) > 3 else 'system'} is configured to operate in a federated learning environment, wherein the processing is distributed across a plurality of nodes without centralizing the input data.\n\n" output += "=" * 60 + "\n\n" output += "NEXT STEPS:\n" output += "1. Review and refine claims with a registered patent attorney\n" output += "2. File a Provisional Patent Application (PPA) to establish priority date (~$320 USPTO fee)\n" output += "3. Conduct prior art search using the check_prior_art tool\n" output += "4. Track this invention using the track_portfolio tool\n" output += "5. File non-provisional within 12 months of provisional\n" return output # --------------------------------------------------------------------------- # Tool 4: check_prior_art # --------------------------------------------------------------------------- @mcp.tool def check_prior_art( invention_description: str, search_depth: str = "standard", ) -> str: """ Search for prior art that could affect patentability of an invention. Searches USPTO patent database and returns relevant existing patents. Args: invention_description: Description of the invention to search prior art for. search_depth: 'quick' (top 5 results), 'standard' (top 10), 'deep' (top 20). Returns: List of potentially relevant prior art patents with patentability assessment. """ depth_limits = {"quick": 5, "standard": 10, "deep": 20} limit = depth_limits.get(search_depth, 10) # Extract key phrases from description desc = invention_description.strip() # Build multiple search queries for thorough coverage words = re.findall(r'\b[a-zA-Z]{4,}\b', desc.lower()) stop_words = {"that", "this", "with", "from", "have", "been", "will", "when", "where", "which", "their", "them", "they", "using", "used", "each", "into", "through", "about", "some", "also", "then"} key_words = [w for w in words if w not in stop_words] word_freq = {} for w in key_words: word_freq[w] = word_freq.get(w, 0) + 1 top_words = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)[:5] search_query = " ".join(w for w, _ in top_words) output = f"PRIOR ART SEARCH REPORT\n" output += f"Invention Summary: {desc[:150]}...\n" if len(desc) > 150 else f"Invention: {desc}\n" output += f"Search Depth: {search_depth} ({limit} results)\n" output += f"Key Search Terms: {search_query}\n" output += "=" * 60 + "\n\n" # Run patent search patent_results_text = search_patents( query=search_query, limit=limit, ) output += "RELEVANT PRIOR ART FOUND:\n\n" output += patent_results_text output += "\n\n" output += "PATENTABILITY ASSESSMENT:\n" output += "=" * 40 + "\n\n" output += "Based on the search results above, evaluate each patent for:\n\n" output += "1. NOVELTY (35 U.S.C. § 102)\n" output += " - Does any single prior art reference disclose ALL elements of your invention?\n" output += " - If YES → your invention lacks novelty and CANNOT be patented as-is\n" output += " - If NO → proceed to obviousness analysis\n\n" output += "2. NON-OBVIOUSNESS (35 U.S.C. § 103)\n" output += " - Could a person skilled in the field combine multiple references to arrive at your invention?\n" output += " - If YES → your invention may be obvious and difficult to patent\n" output += " - If NO → your invention may be patentable\n\n" output += "3. RECOMMENDED DIFFERENTIATORS:\n" output += " - Identify what makes your invention genuinely different from the prior art\n" output += " - Draft claims to emphasize these differences\n" output += " - Consider provisional filing to lock in your priority date\n\n" output += "IMPORTANT: This is a preliminary search only. A thorough prior art search by a\n" output += "professional patent searcher or attorney is strongly recommended before filing.\n" output += f"\nSearch more: https://patents.google.com/?q={requests.utils.quote(search_query)}\n" output += f"USPTO full text: https://patft.uspto.gov/netahtml/PTO/search-bool.html\n" output += f"Espacenet (global): https://worldwide.espacenet.com/patent/search?q={requests.utils.quote(search_query)}\n" return output # --------------------------------------------------------------------------- # Tool 5: track_portfolio # --------------------------------------------------------------------------- @mcp.tool def track_portfolio( action: str = "list", patent_id: str = "", patent_data: str = "", ) -> str: """ Read and write the Genesis patent portfolio tracker. Tracks patent numbers, filing dates, status, deadlines, and notes. Portfolio stored at E:/genesis-system/data/patent_portfolio.json (no SQLite). Args: action: Operation — 'list' (show all), 'add' (add new patent/application), 'update' (update existing), 'remove' (remove entry), 'deadlines' (show upcoming). patent_id: Identifier for the patent (e.g. 'US10123456' or 'PROV-2026-001'). patent_data: JSON string of patent data when action='add' or 'update'. Required fields for 'add': title, status, filing_date Optional fields: patent_number, assignee, inventors, notes, maintenance_deadlines, annual_fee_due, technology_area Returns: Portfolio status or confirmation of operation. """ portfolio = _load_portfolio() patents = portfolio.get("patents", []) action = action.lower().strip() if action == "list": if not patents: return ( "Patent portfolio is empty.\n\n" "Add patents using: track_portfolio(action='add', patent_id='PROV-2026-001', " "patent_data='{\"title\": \"My Invention\", \"status\": \"provisional\", " "\"filing_date\": \"2026-02-23\"}')" ) output = f"GENESIS PATENT PORTFOLIO\n" output += f"Last Updated: {portfolio.get('last_updated', 'Never')}\n" output += f"Total Patents/Applications: {len(patents)}\n" output += "=" * 60 + "\n\n" for i, p in enumerate(patents, 1): output += f"{i}. {p.get('title', 'Untitled')}\n" output += f" ID: {p.get('patent_id', 'N/A')}\n" output += f" Status: {p.get('status', 'unknown')}\n" output += f" Filing Date: {p.get('filing_date', 'N/A')}\n" if p.get("patent_number"): output += f" Patent Number: {p['patent_number']}\n" if p.get("technology_area"): output += f" Technology: {p['technology_area']}\n" if p.get("notes"): output += f" Notes: {p['notes'][:150]}\n" output += "\n" return output elif action == "add": if not patent_id: return "Error: patent_id is required for action='add'." if not patent_data: return "Error: patent_data (JSON string) is required for action='add'." # Check if already exists existing_ids = [p.get("patent_id", "") for p in patents] if patent_id in existing_ids: return f"Error: Patent ID '{patent_id}' already exists. Use action='update' to modify." try: data = json.loads(patent_data) except json.JSONDecodeError as e: return f"Error: Invalid JSON in patent_data: {e}" if not data.get("title"): return "Error: 'title' field is required in patent_data." if not data.get("status"): return "Error: 'status' field is required in patent_data." if not data.get("filing_date"): return "Error: 'filing_date' field is required in patent_data." new_patent = { "patent_id": patent_id, "added_at": datetime.utcnow().isoformat(), **data, } patents.append(new_patent) portfolio["patents"] = patents if _save_portfolio(portfolio): return ( f"Successfully added '{data['title']}' (ID: {patent_id}) to portfolio.\n" f"Status: {data['status']}\n" f"Filing Date: {data['filing_date']}\n" f"Total portfolio size: {len(patents)} patents/applications." ) return "Error: Failed to save portfolio." elif action == "update": if not patent_id: return "Error: patent_id is required for action='update'." if not patent_data: return "Error: patent_data (JSON string) is required for action='update'." try: update_data = json.loads(patent_data) except json.JSONDecodeError as e: return f"Error: Invalid JSON in patent_data: {e}" for i, p in enumerate(patents): if p.get("patent_id") == patent_id: patents[i] = {**p, **update_data, "updated_at": datetime.utcnow().isoformat()} portfolio["patents"] = patents if _save_portfolio(portfolio): return f"Successfully updated patent '{patent_id}'." return "Error: Failed to save portfolio." return f"Error: Patent ID '{patent_id}' not found in portfolio." elif action == "remove": if not patent_id: return "Error: patent_id is required for action='remove'." original_count = len(patents) patents = [p for p in patents if p.get("patent_id") != patent_id] if len(patents) == original_count: return f"Error: Patent ID '{patent_id}' not found in portfolio." portfolio["patents"] = patents if _save_portfolio(portfolio): return f"Successfully removed '{patent_id}' from portfolio. Remaining: {len(patents)} patents." return "Error: Failed to save portfolio." elif action == "deadlines": if not patents: return "Portfolio is empty — no deadlines to track." output = "UPCOMING PATENT DEADLINES\n" output += "=" * 60 + "\n\n" today = datetime.utcnow() deadline_items = [] for p in patents: # Check provisional 12-month deadline if p.get("status") in ("provisional", "provisional_filed"): filing_date = p.get("filing_date", "") if filing_date: try: fd = datetime.strptime(filing_date[:10], "%Y-%m-%d") deadline = fd + timedelta(days=365) days_left = (deadline - today).days deadline_items.append({ "title": p.get("title", "Unknown"), "id": p.get("patent_id", ""), "deadline_type": "Non-provisional filing deadline (12 months from provisional)", "deadline_date": deadline.strftime("%Y-%m-%d"), "days_left": days_left, "urgency": "CRITICAL" if days_left < 30 else "WARNING" if days_left < 90 else "OK", }) except ValueError: pass # Check maintenance fee deadlines for deadline_entry in p.get("maintenance_deadlines", []): if isinstance(deadline_entry, dict): try: dd = datetime.strptime(str(deadline_entry.get("date", ""))[:10], "%Y-%m-%d") days_left = (dd - today).days if days_left > 0: deadline_items.append({ "title": p.get("title", "Unknown"), "id": p.get("patent_id", ""), "deadline_type": deadline_entry.get("type", "Maintenance fee"), "deadline_date": dd.strftime("%Y-%m-%d"), "days_left": days_left, "urgency": "CRITICAL" if days_left < 30 else "WARNING" if days_left < 90 else "OK", }) except ValueError: pass deadline_items.sort(key=lambda x: x["days_left"]) if not deadline_items: output += "No upcoming deadlines found.\n" output += "Add maintenance_deadlines to your portfolio entries to track them.\n" else: for item in deadline_items: urgency_symbol = "!!! " if item["urgency"] == "CRITICAL" else " ! " if item["urgency"] == "WARNING" else " " output += f"{urgency_symbol}[{item['urgency']}] {item['title']} ({item['id']})\n" output += f" {item['deadline_type']}\n" output += f" Due: {item['deadline_date']} ({item['days_left']} days remaining)\n\n" return output else: return ( f"Error: Unknown action '{action}'. " f"Valid actions: list, add, update, remove, deadlines" ) # --------------------------------------------------------------------------- # Tool 6: generate_ip_memo # --------------------------------------------------------------------------- @mcp.tool def generate_ip_memo( invention_description: str, company_name: str = "Genesis", market_context: str = "", competitors: str = "", ) -> str: """ Generate an IP strategy memo for an invention — what to patent, what to keep as trade secret, what to publish defensively, and recommended timeline. Args: invention_description: Detailed description of the invention. company_name: Name of the company (default: Genesis). market_context: Description of the market, problem being solved, and potential customers (optional but improves memo quality). competitors: Comma-separated list of key competitors to address (optional). Returns: Full IP strategy memo with recommendations and action items. """ if not invention_description.strip(): return "Error: invention_description is required." now = datetime.utcnow() competitor_list = [c.strip() for c in competitors.split(",") if c.strip()] if competitors else [] memo = f"IP STRATEGY MEMO\n" memo += "=" * 70 + "\n" memo += f"DATE: {now.strftime('%B %d, %Y')}\n" memo += f"COMPANY: {company_name}\n" memo += f"PREPARED BY: Genesis Patent MCP\n" memo += f"STATUS: DRAFT — For review by registered patent counsel\n" memo += "=" * 70 + "\n\n" memo += "EXECUTIVE SUMMARY\n" memo += "-" * 40 + "\n" desc_preview = invention_description[:300] + "..." if len(invention_description) > 300 else invention_description memo += f"{desc_preview}\n\n" if market_context: memo += f"Market Context: {market_context[:300]}\n\n" memo += "IP PROTECTION STRATEGY\n" memo += "-" * 40 + "\n\n" memo += "1. WHAT TO PATENT\n" memo += " Strong patent candidates are novel, non-obvious technical processes\n" memo += " or systems that provide competitive advantage and are difficult to\n" memo += " independently discover through reverse engineering.\n\n" memo += " Recommended for patent protection:\n" memo += f" - The core technical method described in the invention\n" memo += f" - Any novel data structures or processing architectures\n" memo += f" - Unique user interaction paradigms (if any)\n" memo += f" - System integration methods that provide competitive advantage\n\n" memo += " Filing strategy:\n" memo += " a) File a Provisional Patent Application (PPA) IMMEDIATELY\n" memo += " Cost: ~$320 (micro entity) to $800 (large entity)\n" memo += " Benefit: Locks in priority date for 12 months\n" memo += " Timeline: Can be done in 1-2 weeks\n\n" memo += " b) File Non-Provisional Utility Patent within 12 months\n" memo += " Cost: ~$800-$1,600 (micro entity fees) + attorney ~$5,000-$15,000\n" memo += " Prosecution: 2-4 years average\n" memo += " Benefit: Up to 20 years protection from non-provisional filing date\n\n" memo += "2. WHAT TO KEEP AS TRADE SECRET\n" memo += " Trade secrets are appropriate for elements that:\n" memo += " (a) Cannot be discovered through reverse engineering\n" memo += " (b) Would take too long to patent vs. rate of technology change\n" memo += " (c) Are best protected by secrecy rather than time-limited patent monopoly\n\n" memo += " Candidates for trade secret protection:\n" memo += f" - Specific training data, datasets, or data pipelines\n" memo += f" - Proprietary model weights and fine-tuning approaches\n" memo += f" - Business processes and operational workflows\n" memo += f" - Customer acquisition and pricing strategy specifics\n\n" memo += " Requirements: Must maintain reasonable secrecy (NDAs, access controls)\n\n" memo += "3. WHAT TO PUBLISH DEFENSIVELY\n" memo += " Defensive publication prevents competitors from patenting the same\n" memo += " technology without disclosing your own IP. Use for:\n\n" memo += " - Technical approaches you will NOT pursue but don't want competitors to patent\n" memo += " - Prior art you want to establish in the public record\n" memo += " - Open-source components that underpin your solution\n\n" memo += " Options: IP.com, ArXiv.org, defensive publication journals\n\n" if competitor_list: memo += "4. COMPETITOR ANALYSIS\n" memo += f" Key competitors identified: {', '.join(competitor_list)}\n\n" memo += " Recommended actions:\n" for comp in competitor_list: memo += f" - Search {comp}'s patent portfolio using monitor_competitor_patents('{comp}')\n" memo += " - Map competitor IP to identify freedom-to-operate risks\n" memo += " - Identify white spaces in competitor patent coverage\n\n" memo += "5. RECOMMENDED TIMELINE\n" memo += "-" * 40 + "\n" d0 = now d1 = d0 + timedelta(weeks=2) d2 = d0 + timedelta(weeks=8) d3 = d0 + timedelta(weeks=16) d4 = d0 + timedelta(days=365) d5 = d0 + timedelta(days=365 + 180) memo += f" TODAY ({d0.strftime('%Y-%m-%d')}): Document invention fully. Sign and date invention disclosure.\n" memo += f" {d1.strftime('%Y-%m-%d')}: File Provisional Patent Application.\n" memo += f" {d2.strftime('%Y-%m-%d')}: Complete prior art search (professional).\n" memo += f" {d3.strftime('%Y-%m-%d')}: Draft non-provisional claims with attorney.\n" memo += f" {d4.strftime('%Y-%m-%d')}: DEADLINE — File non-provisional (12 months from provisional).\n" memo += f" {d5.strftime('%Y-%m-%d')}: Respond to first USPTO office action (if any).\n\n" memo += "6. COST ESTIMATE\n" memo += "-" * 40 + "\n" memo += " Provisional Application: $320 - $800 (USPTO fees) + $500-$2,500 (attorney)\n" memo += " Non-Provisional Utility: $800 - $1,600 (USPTO fees) + $5,000-$15,000 (attorney)\n" memo += " PCT International Filing: $4,000 - $8,000 (fees) + $5,000-$10,000 (attorney)\n" memo += " Maintenance Fees (20 years): ~$12,000 total (3.5, 7.5, 11.5 year anniversaries)\n" memo += " TOTAL (US only estimate): $7,000 - $25,000 over patent lifetime\n\n" memo += "7. RISK ASSESSMENT\n" memo += "-" * 40 + "\n" memo += " - NOT filing: Risk of competitors patenting similar technology\n" memo += " - Filing too broadly: Increased invalidity risk, higher prosecution cost\n" memo += " - Filing too narrowly: Easy for competitors to design around\n" memo += " - Delay: Every day without a priority date is a risk\n\n" memo += "ACTION ITEMS\n" memo += "-" * 40 + "\n" memo += " [ ] Complete formal Invention Disclosure Form (IDF)\n" memo += " [ ] Have all inventors sign and date the IDF\n" memo += " [ ] Engage a registered patent attorney for PPA filing\n" memo += " [ ] Run prior art search using check_prior_art tool\n" memo += " [ ] Add to patent portfolio tracker using track_portfolio tool\n" memo += " [ ] Set calendar reminder for 11-month PPA deadline\n\n" memo += "=" * 70 + "\n" memo += "DISCLAIMER: This memo is generated by AI for informational purposes only\n" memo += "and does not constitute legal advice. Consult a registered patent attorney\n" memo += "(USPTO registration required) before making any IP decisions.\n" return memo # --------------------------------------------------------------------------- # Tool 7: calculate_fees # --------------------------------------------------------------------------- @mcp.tool def calculate_fees( patent_type: str = "utility", entity_size: str = "micro", filing_stage: str = "provisional", claim_count: int = 20, independent_claim_count: int = 3, ) -> str: """ Calculate USPTO patent fees for budget planning. Fee schedule based on USPTO 2024-2026 fee schedule. Args: patent_type: 'utility', 'provisional', 'design', 'plant', 'continuation'. entity_size: 'micro' (≤$229K gross income + small entity criteria), 'small' (≤500 employees or qualifying nonprofit), 'large' (everything else). filing_stage: 'provisional', 'nonprovisional', 'maintenance_3.5yr', 'maintenance_7.5yr', 'maintenance_11.5yr', 'pct_national_phase'. claim_count: Total number of claims (default 20 — included free). independent_claim_count: Number of independent claims (default 3 — included free). Returns: Itemized fee breakdown and total estimated USPTO cost. """ # USPTO Fee Schedule (2025 values — verify current fees at fees.uspto.gov) # Multipliers: micro = 0.2, small = 0.4, large = 1.0 ENTITY_MULTIPLIERS = {"micro": 0.20, "small": 0.40, "large": 1.00} multiplier = ENTITY_MULTIPLIERS.get(entity_size.lower(), 1.00) # Base large-entity fees (USD) BASE_FEES = { "provisional_basic": 320, "utility_basic_filing": 320, "utility_basic_search": 700, "utility_basic_examination": 800, "design_basic_filing": 240, "design_basic_search": 160, "design_basic_examination": 620, "plant_basic_filing": 250, "plant_basic_search": 430, "plant_basic_examination": 680, "excess_claim_total_over_20": 100, # per claim over 20 "excess_independent_over_3": 480, # per independent over 3 "multiple_dependent_surcharge": 860, "maintenance_3_5yr": 2000, "maintenance_7_5yr": 3760, "maintenance_11_5yr": 7700, "pct_transmittal": 310, "pct_international_search_epo": 2200, "pct_national_phase_us": 1520, "issue_fee": 1200, "publication_fee": 0, # Currently $0 } fees = [] total = 0 stage = filing_stage.lower().strip() ptype = patent_type.lower().strip() if stage == "provisional": base = BASE_FEES["provisional_basic"] * multiplier fees.append({"item": "Provisional Application Filing Fee", "large_entity": BASE_FEES["provisional_basic"], "your_fee": round(base)}) total += base elif stage == "nonprovisional": if ptype in ("utility", "continuation"): for key, label in [ ("utility_basic_filing", "Basic Filing Fee"), ("utility_basic_search", "Search Fee"), ("utility_basic_examination", "Examination Fee"), ]: fee = BASE_FEES[key] * multiplier fees.append({"item": label, "large_entity": BASE_FEES[key], "your_fee": round(fee)}) total += fee elif ptype == "design": for key, label in [ ("design_basic_filing", "Design Basic Filing Fee"), ("design_basic_search", "Design Search Fee"), ("design_basic_examination", "Design Examination Fee"), ]: fee = BASE_FEES[key] * multiplier fees.append({"item": label, "large_entity": BASE_FEES[key], "your_fee": round(fee)}) total += fee elif ptype == "plant": for key, label in [ ("plant_basic_filing", "Plant Basic Filing Fee"), ("plant_basic_search", "Plant Search Fee"), ("plant_basic_examination", "Plant Examination Fee"), ]: fee = BASE_FEES[key] * multiplier fees.append({"item": label, "large_entity": BASE_FEES[key], "your_fee": round(fee)}) total += fee # Excess claims excess_total = max(0, claim_count - 20) if excess_total > 0: fee = BASE_FEES["excess_claim_total_over_20"] * multiplier * excess_total fees.append({ "item": f"Excess Claims ({excess_total} claims over 20)", "large_entity": BASE_FEES["excess_claim_total_over_20"] * excess_total, "your_fee": round(fee), }) total += fee excess_independent = max(0, independent_claim_count - 3) if excess_independent > 0: fee = BASE_FEES["excess_independent_over_3"] * multiplier * excess_independent fees.append({ "item": f"Excess Independent Claims ({excess_independent} over 3)", "large_entity": BASE_FEES["excess_independent_over_3"] * excess_independent, "your_fee": round(fee), }) total += fee # Issue fee (paid upon allowance) issue_fee = BASE_FEES["issue_fee"] * multiplier fees.append({ "item": "Issue Fee (paid upon allowance)", "large_entity": BASE_FEES["issue_fee"], "your_fee": round(issue_fee), }) total += issue_fee elif stage == "maintenance_3.5yr": fee = BASE_FEES["maintenance_3_5yr"] * multiplier fees.append({"item": "Maintenance Fee (3.5 Year)", "large_entity": BASE_FEES["maintenance_3_5yr"], "your_fee": round(fee)}) total += fee elif stage == "maintenance_7.5yr": fee = BASE_FEES["maintenance_7_5yr"] * multiplier fees.append({"item": "Maintenance Fee (7.5 Year)", "large_entity": BASE_FEES["maintenance_7_5yr"], "your_fee": round(fee)}) total += fee elif stage == "maintenance_11.5yr": fee = BASE_FEES["maintenance_11_5yr"] * multiplier fees.append({"item": "Maintenance Fee (11.5 Year)", "large_entity": BASE_FEES["maintenance_11_5yr"], "your_fee": round(fee)}) total += fee elif stage == "pct_national_phase": for key, label in [ ("pct_transmittal", "PCT Transmittal Fee"), ("pct_national_phase_us", "PCT National Phase Entry (US)"), ]: fee = BASE_FEES[key] * multiplier fees.append({"item": label, "large_entity": BASE_FEES[key], "your_fee": round(fee)}) total += fee # Build output entity_labels = {"micro": "Micro Entity (20% of large entity fees)", "small": "Small Entity (40% of large entity fees)", "large": "Large Entity (full fees)"} output = "USPTO PATENT FEE ESTIMATE\n" output += "=" * 60 + "\n" output += f"Patent Type: {patent_type.upper()}\n" output += f"Filing Stage: {filing_stage}\n" output += f"Entity Size: {entity_labels.get(entity_size.lower(), entity_size)}\n" output += f"Claims: {claim_count} total, {independent_claim_count} independent\n" output += "=" * 60 + "\n\n" output += f"{'ITEM':<45} {'LARGE ENT':>10} {'YOUR FEE':>10}\n" output += "-" * 70 + "\n" for fee in fees: output += f"{fee['item']:<45} ${fee['large_entity']:>9,} ${fee['your_fee']:>9,}\n" output += "-" * 70 + "\n" output += f"{'TOTAL USPTO GOVERNMENT FEES':<45} ${sum(f['large_entity'] for f in fees):>9,} ${round(total):>9,}\n\n" output += "ADDITIONAL COSTS (not included above):\n" output += " Patent Attorney Fees:\n" output += " Provisional preparation: $500 - $2,500\n" output += " Non-provisional drafting: $5,000 - $15,000\n" output += " Prosecution (per OA): $1,500 - $4,000\n" output += " Professional Prior Art Search: $500 - $2,000\n" output += " Translation (PCT, per country): $1,500 - $3,000\n\n" output += "LIFETIME COST ESTIMATE (US utility patent to expiry):\n" output += " Filing (micro entity): ~$700\n" output += " Attorney fees: ~$10,000 - $20,000\n" output += " Maintenance fees (20 years): ~$2,700 (micro entity total)\n" output += " TOTAL ESTIMATE: ~$13,000 - $23,000\n\n" output += "VERIFY CURRENT FEES: https://fees.uspto.gov/\n" output += "MICRO ENTITY CRITERIA: https://www.uspto.gov/patents/basics/micro-entity-status\n" output += "\nNOTE: Fees subject to change. Verify at USPTO before filing.\n" output += "These are government fees only — attorney fees are additional.\n" return output # --------------------------------------------------------------------------- # Tool 8: monitor_competitor_patents # --------------------------------------------------------------------------- @mcp.tool def monitor_competitor_patents( company_name: str, years_back: int = 2, technology_focus: str = "", ) -> str: """ Search for recent patents filed by a competitor company. Monitors competitor IP activity to identify threats and opportunities. Args: company_name: Name of the competitor company to monitor. years_back: How many years back to search (default 2, max 5). technology_focus: Optional technology area to narrow results (e.g. 'machine learning', 'voice recognition', 'privacy'). Returns: List of recent competitor patents with analysis of their IP strategy. """ years_back = min(max(1, years_back), 5) date_from = (datetime.utcnow() - timedelta(days=365 * years_back)).strftime("%Y-%m-%d") query = technology_focus if technology_focus else "artificial intelligence" output = f"COMPETITOR PATENT MONITOR\n" output += f"Company: {company_name}\n" output += f"Period: Last {years_back} year(s) (from {date_from})\n" if technology_focus: output += f"Technology Focus: {technology_focus}\n" output += "=" * 60 + "\n\n" # Search for competitor patents search_result = search_patents( query=query, date_from=date_from, assignee=company_name, limit=15, ) output += "RECENT PATENTS FILED:\n\n" output += search_result output += "\n\n" output += "COMPETITIVE INTELLIGENCE ANALYSIS:\n" output += "=" * 40 + "\n\n" output += f"1. STRATEGIC IMPLICATIONS\n" output += f" - Review each patent above for potential overlap with {company_name} products\n" output += f" - Identify technology areas where {company_name} is building an IP moat\n" output += f" - Note any pending applications that could affect your freedom to operate\n\n" output += f"2. FREEDOM-TO-OPERATE CONSIDERATIONS\n" output += f" - Check if any {company_name} patents cover techniques you are using\n" output += f" - Consult a patent attorney for FTO opinion before product launch\n" output += f" - Consider design-around alternatives for any blocking patents\n\n" output += f"3. IP WHITE SPACE OPPORTUNITIES\n" output += f" - Identify technology areas NOT covered by {company_name}'s patents\n" output += f" - These white spaces represent opportunities for your own patent filings\n\n" output += f"4. RECOMMENDED ACTIONS\n" output += f" [ ] Set up Google Patents alerts for {company_name}: https://patents.google.com/?assignee={requests.utils.quote(company_name)}&sort=new\n" output += f" [ ] Monitor USPTO assignments for {company_name}: https://assignment.uspto.gov/\n" output += f" [ ] Review {company_name}'s pending patent applications (pre-grant publications)\n" output += f" [ ] Brief your patent attorney on competitive landscape\n\n" output += f"MORE RESOURCES:\n" output += f" Google Patents ({company_name}): https://patents.google.com/?assignee={requests.utils.quote(company_name)}&sort=new\n" output += f" USPTO Patent Full Text: https://patft.uspto.gov/netahtml/PTO/search-bool.html\n" output += f" Espacenet Global: https://worldwide.espacenet.com/patent/search?q=an%3D{requests.utils.quote(company_name)}\n" return output # --------------------------------------------------------------------------- # Run server # --------------------------------------------------------------------------- if __name__ == "__main__": import uvicorn logger.info("Starting Genesis Patent MCP Server") logger.info("Port: %s | Transport: %s", PORT, TRANSPORT) logger.info("Portfolio file: %s", PORTFOLIO_FILE) # Build ASGI app with middleware app = mcp.streamable_http_app() if TRANSPORT == "http" else mcp.sse_app() # Wrap with auth middleware from starlette.applications import Starlette from starlette.routing import Route, Mount from starlette.middleware import Middleware as StarletteMiddleware # Apply Bearer token middleware app_with_auth = BearerTokenMiddleware(app, AUTH_TOKEN) uvicorn.run( app_with_auth, host=HOST, port=PORT, log_level="info", )