M4

Instructions

Implementation Reference

• src/m4/core/tools/notes.py:226-316 (handler)

  The ListPatientNotesTool class implements the core logic for the list_patient_notes tool. It queries the database for note metadata (IDs, types, lengths, previews) for a given subject_id across specified note types without loading full text.
  class ListPatientNotesTool: """Tool for listing available notes for a patient. Returns metadata only (note IDs, types, lengths) - not full text. Use this to discover what notes exist before retrieving them. """ name = "list_patient_notes" description = ( "List available clinical notes for a patient by subject_id. " "Returns note metadata (IDs, types, lengths) without full text. " "Use get_note(note_id) to retrieve specific notes." ) input_model = ListPatientNotesInput output_model = ToolOutput required_modalities: frozenset[Modality] = frozenset({Modality.NOTES}) supported_datasets: frozenset[str] | None = None def invoke( self, dataset: DatasetDefinition, params: ListPatientNotesInput ) -> ToolOutput: """List notes for a patient without returning full text.""" backend = get_backend() backend_info = backend.get_backend_info(dataset) tables_to_query = self._get_tables_for_type(params.note_type) if not tables_to_query: return ToolOutput( result=f"{backend_info}\n**Error:** Invalid note_type '{params.note_type}'. " f"Use 'discharge', 'radiology', or 'all'." ) results = [] for table in tables_to_query: # Query for metadata only - explicitly exclude full text sql = f""" SELECT note_id, subject_id, '{table}' as note_type, LENGTH(text) as note_length, LEFT(text, 100) as preview FROM {table} WHERE subject_id = {params.subject_id} LIMIT {params.limit} """ try: result = backend.execute_query(sql, dataset) if result.success and result.data: results.append(f"\n**{table.upper()} NOTES:**\n{result.data}") except Exception as e: results.append(f"\n**{table.upper()}:** Error - {e}") if not results or all("no rows" in r.lower() for r in results): return ToolOutput( result=f"{backend_info}\n**No notes found** for subject_id {params.subject_id}.\n\n" f"**Tip:** Verify the subject_id exists in the related MIMIC-IV dataset." ) output = ( f"{backend_info}\n" f"**Notes for subject_id {params.subject_id}:**\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:48-54 (schema)

  Dataclass defining the input parameters for the list_patient_notes tool: subject_id (required), note_type (default 'all'), limit (default 20).
  class ListPatientNotesInput(ToolInput): """Input for list_patient_notes tool.""" subject_id: int note_type: str = "all" # discharge, radiology, or all limit: int = 20
• src/m4/core/tools/__init__.py:37-74 (registration)

  The init_tools() function registers the ListPatientNotesTool (along with other tools) into the ToolRegistry, making it available for use.
  def init_tools() -> None: """Initialize and register all available tools. This function registers all tool classes with the ToolRegistry. It is idempotent and thread-safe - calling it multiple times or from multiple threads has no additional effect. This should be called during application startup, before the MCP server begins accepting requests. Example: from m4.core.tools import init_tools init_tools() # Register all tools """ global _tools_initialized with _tools_lock: # Check if already initialized AND tools are registered # (handles case where registry was reset but flag is still True) if _tools_initialized and ToolRegistry.list_all(): return # Register management tools (always available) ToolRegistry.register(ListDatasetsTool()) ToolRegistry.register(SetDatasetTool()) # Register tabular data tools ToolRegistry.register(GetDatabaseSchemaTool()) ToolRegistry.register(GetTableInfoTool()) ToolRegistry.register(ExecuteQueryTool()) # Register clinical notes tools ToolRegistry.register(SearchNotesTool()) ToolRegistry.register(GetNoteTool()) ToolRegistry.register(ListPatientNotesTool()) _tools_initialized = True
• src/m4/mcp_server.py:258-293 (handler)

  MCP server adapter: FastMCP-decorated function that handles the list_patient_notes tool call, checks compatibility, retrieves the tool from registry, and invokes it with parsed inputs.
  def list_patient_notes( subject_id: int, note_type: str = "all", limit: int = 20, ) -> str: """📋 List available clinical notes for a patient. Returns note metadata (IDs, types, lengths) without full text. Use get_note(note_id) to retrieve specific notes. **Cross-dataset tip:** Get subject_id from MIMIC-IV queries, then use it here to find related clinical notes. Args: subject_id: Patient identifier (same as in MIMIC-IV). note_type: Type of notes to list ('discharge', 'radiology', or 'all'). limit: Maximum notes to return (default: 20). Returns: List of available notes with metadata for the patient. """ dataset = DatasetRegistry.get_active() result = _tool_selector.check_compatibility("list_patient_notes", dataset) if not result.compatible: return result.error_message tool = ToolRegistry.get("list_patient_notes") return tool.invoke( dataset, ListPatientNotesInput( subject_id=subject_id, note_type=note_type, limit=limit, ), ).result