DataBeak

Instructions

Implementation Reference

• src/databeak/servers/io_server.py:431-456 (handler)

  The handler function for the 'get_session_info' tool. Retrieves the current session ID from the MCP context, fetches the session object, calls get_info() on it, and returns a formatted SessionInfoResult.

  async def get_session_info(
      ctx: Annotated[Context, Field(description="FastMCP context for session access")],
  ) -> SessionInfoResult:
      """Get comprehensive information about a specific session.
  
      Returns session metadata, data status, and configuration. Essential for session management and
      workflow coordination.
      """
      # Get session_id from FastMCP context
      session_id = ctx.session_id
  
      session = get_session_only(session_id)
  
      await ctx.info(f"Retrieved info for session {session_id}")
  
      # Get comprehensive session information
      info = session.get_info()
  
      return SessionInfoResult(
          created_at=info.created_at.isoformat(),
          last_modified=info.last_accessed.isoformat(),
          data_loaded=session.has_data(),
          row_count=info.row_count if session.has_data() else None,
          column_count=info.column_count if session.has_data() else None,
      )

• src/databeak/servers/io_server.py:472-472 (registration)

  Registration of the 'get_session_info' tool on the io_server FastMCP instance.

  io_server.tool(name="load_csv_from_content")(load_csv_from_content)

• src/databeak/servers/io_server.py:116-125 (schema)

  Pydantic model defining the output schema of the get_session_info tool.

  class SessionInfoResult(BaseToolResponse):
      """Response model for session information."""
  
      created_at: str = Field(description="Creation timestamp (ISO format)")
      last_modified: str = Field(description="Last modification timestamp (ISO format)")
      data_loaded: bool = Field(description="Whether session has data loaded")
      row_count: int | None = Field(None, description="Number of rows if data loaded")
      column_count: int | None = Field(None, description="Number of columns if data loaded")

• src/databeak/models/data_models.py:131-143 (schema)

  Core Pydantic model used by session.get_info() to structure session metadata, which is then adapted into SessionInfoResult.

  class SessionInfo(BaseModel):
      """Information about a data session."""
  
      session_id: str = Field(..., description="Unique session identifier")
      created_at: datetime = Field(..., description="Session creation time")
      last_accessed: datetime = Field(..., description="Last access time")
      row_count: int = Field(..., description="Number of rows in dataset")
      column_count: int = Field(..., description="Number of columns")
      columns: list[str] = Field(..., description="Column names")
      memory_usage_mb: float = Field(..., description="Memory usage in MB")
      operations_count: int = Field(default=0, description="Number of operations performed")
      file_path: str | None = Field(default=None, description="Source file path")

• src/databeak/core/session.py:132-147 (helper)

  DatabeakSession.get_info() method that aggregates data shape, columns, memory usage from _data_session and lifecycle info to create a SessionInfo object.

  """Get session information."""
  data_info = self._data_session.get_data_info()
  lifecycle_info = self.lifecycle.get_lifecycle_info()
  
  return SessionInfo(
      session_id=self.session_id,
      created_at=lifecycle_info["created_at"],
      last_accessed=lifecycle_info["last_accessed"],
      row_count=data_info["shape"][0],
      column_count=data_info["shape"][1],
      columns=data_info["columns"],
      memory_usage_mb=data_info["memory_usage_mb"],
      operations_count=0,  # No longer tracking operations (simplified architecture)
      file_path=data_info["file_path"],
  )