M4

Instructions

Implementation Reference

• src/m4/core/tools/notes.py:176-216 (handler)

  The invoke method of the GetNoteTool class, which implements the core logic of the 'get_note' tool by querying the database for the specified note ID across discharge and radiology tables, handling truncation if specified, and returning the note text or an error.

  def invoke(self, dataset: DatasetDefinition, params: GetNoteInput) -> ToolOutput:
      """Retrieve a single note by ID."""
      backend = get_backend()
      backend_info = backend.get_backend_info(dataset)
  
      # Note IDs contain the note type (e.g., "10000032_DS-1" for discharge)
      note_id = params.note_id.replace("'", "''")
  
      # Try both tables since we may not know which one contains the note
      for table in ["discharge", "radiology"]:
          sql = f"""
              SELECT
                  note_id,
                  subject_id,
                  text,
                  LENGTH(text) as note_length
              FROM {table}
              WHERE note_id = '{note_id}'
              LIMIT 1
          """
  
          try:
              result = backend.execute_query(sql, dataset)
              if result.success and result.data and "note_id" in result.data.lower():
                  # Found the note
                  # Optionally truncate if max_length specified
                  if params.max_length and len(result.data) > params.max_length:
                      truncated_result = result.data[: params.max_length]
                      return ToolOutput(
                          result=f"{backend_info}\n**Note (truncated to {params.max_length} chars):**\n{truncated_result}\n\n[...truncated...]"
                      )
                  return ToolOutput(result=f"{backend_info}\n{result.data}")
          except Exception:
              continue
  
      return ToolOutput(
          result=f"{backend_info}\n**Error:** Note '{params.note_id}' not found. "
          f"Use `list_patient_notes(subject_id)` or `search_notes(query)` "
          f"to find valid note IDs."
      )

• src/m4/core/tools/notes.py:40-45 (schema)

  Dataclass defining the input schema for the get_note tool, including required note_id and optional max_length for truncation.

  class GetNoteInput(ToolInput):
      """Input for get_note tool."""
  
      note_id: str
      max_length: int | None = None  # Optional truncation

• src/m4/core/tools/__init__.py:70-70 (registration)

  Registration of the GetNoteTool instance in the ToolRegistry during tool initialization.

  ToolRegistry.register(GetNoteTool())

• src/m4/core/tools/__init__.py:18-22 (registration)

  Import of GetNoteTool from notes.py module for use in registration.

  from m4.core.tools.notes import (
      GetNoteTool,
      ListPatientNotesTool,
      SearchNotesTool,
  )

• src/m4/core/tools/notes.py:158-224 (handler)

  Full GetNoteTool class definition, including name 'get_note', description, input/output models, required modalities, invoke handler, and compatibility check.

  class GetNoteTool:
      """Tool for retrieving a single clinical note by ID.
  
      Returns the full note text. Use with caution as notes can be long.
      """
  
      name = "get_note"
      description = (
          "Retrieve full text of a specific clinical note by note_id. "
          "Notes can be very long - consider using search_notes() first "
          "to find relevant notes."
      )
      input_model = GetNoteInput
      output_model = ToolOutput
  
      required_modalities: frozenset[Modality] = frozenset({Modality.NOTES})
      supported_datasets: frozenset[str] | None = None
  
      def invoke(self, dataset: DatasetDefinition, params: GetNoteInput) -> ToolOutput:
          """Retrieve a single note by ID."""
          backend = get_backend()
          backend_info = backend.get_backend_info(dataset)
  
          # Note IDs contain the note type (e.g., "10000032_DS-1" for discharge)
          note_id = params.note_id.replace("'", "''")
  
          # Try both tables since we may not know which one contains the note
          for table in ["discharge", "radiology"]:
              sql = f"""
                  SELECT
                      note_id,
                      subject_id,
                      text,
                      LENGTH(text) as note_length
                  FROM {table}
                  WHERE note_id = '{note_id}'
                  LIMIT 1
              """
  
              try:
                  result = backend.execute_query(sql, dataset)
                  if result.success and result.data and "note_id" in result.data.lower():
                      # Found the note
                      # Optionally truncate if max_length specified
                      if params.max_length and len(result.data) > params.max_length:
                          truncated_result = result.data[: params.max_length]
                          return ToolOutput(
                              result=f"{backend_info}\n**Note (truncated to {params.max_length} chars):**\n{truncated_result}\n\n[...truncated...]"
                          )
                      return ToolOutput(result=f"{backend_info}\n{result.data}")
              except Exception:
                  continue
  
          return ToolOutput(
              result=f"{backend_info}\n**Error:** Note '{params.note_id}' not found. "
              f"Use `list_patient_notes(subject_id)` or `search_notes(query)` "
              f"to find valid note IDs."
          )
  
      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