M4

Instructions

Implementation Reference

• src/m4/core/tools/notes.py:57-156 (handler)

  Core handler implementation: SearchNotesTool class with invoke() method that performs full-text search on clinical notes using SQL, extracts snippets around matches, and formats results.
  class SearchNotesTool: """Tool for full-text search across clinical notes. Returns snippets around matches to prevent context overflow. Use get_note() to retrieve full text of specific notes. """ name = "search_notes" description = ( "Search clinical notes by keyword. Returns snippets, not full text. " "Use get_note() to retrieve full content of a specific note." ) input_model = SearchNotesInput output_model = ToolOutput required_modalities: frozenset[Modality] = frozenset({Modality.NOTES}) supported_datasets: frozenset[str] | None = None def invoke( self, dataset: DatasetDefinition, params: SearchNotesInput ) -> ToolOutput: """Search notes and return snippets around matches.""" backend = get_backend() backend_info = backend.get_backend_info(dataset) # Determine which tables to search tables_to_search = self._get_tables_for_type(params.note_type) if not tables_to_search: return ToolOutput( result=f"{backend_info}\n**Error:** Invalid note_type '{params.note_type}'. " f"Use 'discharge', 'radiology', or 'all'." ) results = [] search_term = params.query.replace("'", "''") # Escape single quotes for table in tables_to_search: # Build search query with snippet extraction # Using LIKE for basic search - could be enhanced with full-text search sql = f""" SELECT note_id, subject_id, CASE WHEN POSITION(LOWER('{search_term}') IN LOWER(text)) > 0 THEN SUBSTRING( text, GREATEST(1, POSITION(LOWER('{search_term}') IN LOWER(text)) - {params.snippet_length // 2}), {params.snippet_length} ) ELSE LEFT(text, {params.snippet_length}) END as snippet, LENGTH(text) as note_length FROM {table} WHERE LOWER(text) LIKE '%{search_term.lower()}%' LIMIT {params.limit} """ try: result = backend.execute_query(sql, dataset) if result.success and result.data: results.append(f"\n**{table.upper()}:**\n{result.data}") except Exception as e: results.append(f"\n**{table.upper()}:** Error - {e}") if not results: return ToolOutput( result=f"{backend_info}\n**No matches found** for '{params.query}' " f"in {', '.join(tables_to_search)}." ) output = ( f"{backend_info}\n" f"**Search:** '{params.query}' (showing snippets of ~{params.snippet_length} chars)\n" f"{''.join(results)}\n\n" f"**Tip:** Use `get_note(note_id)` to retrieve full text of a specific note." ) return ToolOutput(result=output) def _get_tables_for_type(self, note_type: str) -> list[str]: """Get table names for a note type.""" note_type = note_type.lower() if note_type == "discharge": return ["discharge"] elif note_type == "radiology": return ["radiology"] elif note_type == "all": return ["discharge", "radiology"] return [] def is_compatible(self, dataset: DatasetDefinition) -> bool: """Check compatibility.""" if self.supported_datasets and dataset.name not in self.supported_datasets: return False if not self.required_modalities.issubset(dataset.modalities): return False return True
• src/m4/core/tools/notes.py:30-37 (schema)

  Input schema definition: SearchNotesInput dataclass specifying parameters for the search_notes tool.
  class SearchNotesInput(ToolInput): """Input for search_notes tool.""" query: str note_type: str = "all" # discharge, radiology, or all limit: int = 5 snippet_length: int = 300
• src/m4/core/tools/__init__.py:69-69 (registration)

  Registration of the SearchNotesTool instance in the central ToolRegistry.
  ToolRegistry.register(SearchNotesTool())
• src/m4/mcp_server.py:187-225 (handler)

  MCP protocol thin adapter/handler: Delegates search_notes calls to the registered SearchNotesTool after compatibility check.
  def search_notes( query: str, note_type: str = "all", limit: int = 5, snippet_length: int = 300, ) -> str: """🔍 Search clinical notes by keyword. Returns snippets around matches to prevent context overflow. Use get_note() to retrieve full text of specific notes. **Note types:** 'discharge' (summaries), 'radiology' (reports), or 'all' Args: query: Search term to find in notes. note_type: Type of notes to search ('discharge', 'radiology', or 'all'). limit: Maximum number of results per note type (default: 5). snippet_length: Characters of context around matches (default: 300). Returns: Matching snippets with note IDs for follow-up retrieval. """ dataset = DatasetRegistry.get_active() result = _tool_selector.check_compatibility("search_notes", dataset) if not result.compatible: return result.error_message tool = ToolRegistry.get("search_notes") return tool.invoke( dataset, SearchNotesInput( query=query, note_type=note_type, limit=limit, snippet_length=snippet_length, ), ).result
• src/m4/core/tools/__init__.py:18-22 (registration)

  Import of SearchNotesTool class required for registration in tool init.
  from m4.core.tools.notes import ( GetNoteTool, ListPatientNotesTool, SearchNotesTool, )