Trade Surveillance Support MCP Server

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• trade_surveillance_mcp/server.py:236-278 (handler)

  The main handler function for the 'search_java_code' tool. Decorated with @mcp.tool() for automatic registration. Takes search keywords and optional code directory, scans if necessary using metadata_index.scan_java_classes(), performs search via metadata_index.search(), and returns matching Java classes with metadata.
  @mcp.tool() async def search_java_code( search_keywords: str, code_directory: str = "./src" ) -> dict[str, Any]: """ Search for Java classes using metadata keywords instead of file paths. This tool searches through indexed Java files by their javadoc metadata annotations. Java classes should include metadata in javadoc comments like: /** * @keywords trade, settlement, report_generator * @type report_engine * @description Generates daily settlement reports */ Args: search_keywords: Keywords to search for (e.g., "report generator", "trade processor") code_directory: Path to the directory containing Java source files (used for initial scan) Returns: A dictionary containing matching Java files with their metadata and methods """ logger.info(f"Searching Java code for keywords: {search_keywords}") # Scan directory if index is empty if not metadata_index.index.get("java_classes"): logger.info(f"Scanning Java classes in: {code_directory}") metadata_index.scan_java_classes(code_directory) # Search by keywords matches = metadata_index.search(search_keywords, file_type="java") result = { "status": "success", "search_keywords": search_keywords, "matches_found": len(matches), "java_classes": matches } logger.info(f"Found {len(matches)} Java class matches") return result
• trade_surveillance_mcp/metadata_index.py:156-204 (helper)

  Core search method in MetadataIndex class, called by the handler with file_type='java'. Matches query terms against keywords, type, description, and file_name in the index.
  def search(self, query: str, file_type: str = "all") -> list[dict[str, Any]]: """ Search the index by keywords, type, or description. Args: query: Search terms (space or comma separated) file_type: "sql", "java", or "all" Returns: List of matching files with their metadata """ query_terms = [term.strip().lower() for term in re.split(r'[,\s]+', query)] results = [] # Search SQL configs if file_type in ("sql", "all"): for file_name, metadata in self.index.get("sql_configs", {}).items(): if self._matches_query(metadata, query_terms): results.append({ "type": "sql_config", "file": file_name, **metadata }) # Search Java classes if file_type in ("java", "all"): for file_name, metadata in self.index.get("java_classes", {}).items(): if self._matches_query(metadata, query_terms): results.append({ "type": "java_class", "file": file_name, **metadata }) return results def _matches_query(self, metadata: dict[str, Any], query_terms: list[str]) -> bool: """Check if metadata matches any query terms.""" # Combine all searchable fields searchable = " ".join([ " ".join(metadata.get("keywords", [])), metadata.get("type", ""), metadata.get("description", ""), metadata.get("file_name", "") ]).lower() # Match if any query term is found return any(term in searchable for term in query_terms)
• trade_surveillance_mcp/metadata_index.py:90-113 (helper)

  Scans Java source files in the given directory, extracts metadata using _extract_java_metadata, and stores in the index. Called by handler if index empty.
  def scan_java_classes(self, code_dir: str) -> dict[str, Any]: """ Scan Java files and extract metadata from javadoc comments. Looks for annotations like: /** * @keywords trade, settlement, report_generator * @type report_engine * @description Generates daily settlement reports */ """ code_path = Path(code_dir) if not code_path.exists(): return {} classes = {} for java_file in code_path.rglob("*.java"): metadata = self._extract_java_metadata(java_file) if metadata: classes[str(java_file.relative_to(code_path))] = metadata self.index["java_classes"] = classes self._save_index() return classes
• trade_surveillance_mcp/metadata_index.py:115-154 (helper)

  Private method that parses individual Java files to extract Javadoc metadata annotations (@keywords, @type, @description) and public methods, used during scanning.
  def _extract_java_metadata(self, file_path: Path) -> dict[str, Any]: """Extract metadata from Java file javadoc comments.""" metadata = { "file_path": str(file_path), "file_name": file_path.name, "class_name": file_path.stem, "keywords": [], "type": None, "description": None, "methods": [] } try: with open(file_path, 'r') as f: content = f.read() # Extract @keywords keywords_match = re.search(r'\*\s*@keywords[:\s]+([^\n]+)', content, re.IGNORECASE) if keywords_match: keywords = [k.strip() for k in keywords_match.group(1).split(',')] metadata["keywords"] = keywords # Extract @type type_match = re.search(r'\*\s*@type[:\s]+([^\n]+)', content, re.IGNORECASE) if type_match: metadata["type"] = type_match.group(1).strip() # Extract @description desc_match = re.search(r'\*\s*@description[:\s]+([^\n]+)', content, re.IGNORECASE) if desc_match: metadata["description"] = desc_match.group(1).strip() # Extract public methods method_pattern = r'public\s+(?:static\s+)?[\w<>\[\]]+\s+(\w+)\s*\(' methods = re.findall(method_pattern, content) metadata["methods"] = methods return metadata if metadata["keywords"] or metadata["type"] else None except Exception: return None