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,
  )