notebook_read
Extract code, text, and outputs from Jupyter notebook files to analyze data science projects and computational workflows.
Instructions
Reads a Jupyter notebook (.ipynb file) and returns all of the cells with their outputs. Jupyter notebooks are interactive documents that combine code, text, and visualizations, commonly used for data analysis and scientific computing. The notebook_path parameter must be an absolute path, not a relative path.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| notebook_path | Yes | The absolute path to the Jupyter notebook file to read (must be absolute, not relative) |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- Core handler implementation that performs path validation, reads the .ipynb file, parses it using JSON, processes cells and outputs via base class methods, formats the content, and returns the formatted string.
@override async def call( self, ctx: MCPContext, **params: Unpack[NotebookReadToolParams], ) -> str: """Execute the tool with the given parameters. Args: ctx: MCP context **params: Tool parameters Returns: Tool result """ tool_ctx = self.create_tool_context(ctx) self.set_tool_context_info(tool_ctx) # Extract parameters notebook_path: NotebookPath = params["notebook_path"] # Validate path parameter path_validation = self.validate_path(notebook_path) if path_validation.is_error: await tool_ctx.error(path_validation.error_message) return f"Error: {path_validation.error_message}" await tool_ctx.info(f"Reading notebook: {notebook_path}") # Check if path is allowed if not self.is_path_allowed(notebook_path): await tool_ctx.error( f"Access denied - path outside allowed directories: {notebook_path}" ) return f"Error: Access denied - path outside allowed directories: {notebook_path}" try: file_path = Path(notebook_path) if not file_path.exists(): await tool_ctx.error(f"File does not exist: {notebook_path}") return f"Error: File does not exist: {notebook_path}" if not file_path.is_file(): await tool_ctx.error(f"Path is not a file: {notebook_path}") return f"Error: Path is not a file: {notebook_path}" # Check file extension if file_path.suffix.lower() != ".ipynb": await tool_ctx.error(f"File is not a Jupyter notebook: {notebook_path}") return f"Error: File is not a Jupyter notebook: {notebook_path}" # Read and parse the notebook try: # This will read the file, so we don't need to read it separately _, processed_cells = await self.parse_notebook(file_path) # Format the notebook content as a readable string result = self.format_notebook_cells(processed_cells) await tool_ctx.info( f"Successfully read notebook: {notebook_path} ({len(processed_cells)} cells)" ) return result except json.JSONDecodeError: await tool_ctx.error(f"Invalid notebook format: {notebook_path}") return f"Error: Invalid notebook format: {notebook_path}" except UnicodeDecodeError: await tool_ctx.error(f"Cannot read notebook file: {notebook_path}") return f"Error: Cannot read notebook file: {notebook_path}" except Exception as e: await tool_ctx.error(f"Error reading notebook: {str(e)}") return f"Error reading notebook: {str(e)}" - Input schema definition: Annotated NotebookPath type and TypedDict for tool parameters.
NotebookPath = Annotated[ str, Field( description="The absolute path to the Jupyter notebook file to read (must be absolute, not relative)", ), ] class NotebookReadToolParams(TypedDict): """Parameters for the NotebookReadTool. Attributes: notebook_path: The absolute path to the Jupyter notebook file to read (must be absolute, not relative) """ notebook_path: NotebookPath - mcp_claude_code/tools/jupyter/notebook_read.py:134-153 (registration)Tool registration method that defines the 'notebook_read' wrapper function with input schema and decorates it with @mcp_server.tool using the tool's name and description.
def register(self, mcp_server: FastMCP) -> None: """Register this read notebook tool with the MCP server. Creates a wrapper function with explicitly defined parameters that match the tool's parameter schema and registers it with the MCP server. Args: mcp_server: The FastMCP server instance """ tool_self = self # Create a reference to self for use in the closure @mcp_server.tool(name=self.name, description=self.description) async def notebook_read( ctx: MCPContext, notebook_path: NotebookPath, ) -> str: ctx = get_context() return await tool_self.call(ctx, notebook_path=notebook_path) - mcp_claude_code/tools/jupyter/__init__.py:48-51 (registration)Instantiation of NotebookReadTool in get_jupyter_tools function, which is called during registration.
return [ NotebookReadTool(permission_manager), NoteBookEditTool(permission_manager), ] - mcp_claude_code/tools/__init__.py:57-60 (registration)Top-level call to register_jupyter_tools during register_all_tools, which instantiates and registers the notebook_read tool.
# Register all jupyter tools jupyter_tools = register_jupyter_tools(mcp_server, permission_manager) for tool in jupyter_tools: all_tools[tool.name] = tool - Key helper method in base class that parses .ipynb JSON, extracts cells, processes outputs (text, images, errors, streams), cleans ANSI codes, and returns processed cells used by the handler.
async def parse_notebook( self, file_path: Path ) -> tuple[dict[str, Any], list[NotebookCellSource]]: """Parse a Jupyter notebook file. Args: file_path: Path to the notebook file Returns: Tuple of (notebook_data, processed_cells) """ with open(file_path, "r", encoding="utf-8") as f: content = f.read() notebook = json.loads(content) # Get notebook language language = ( notebook.get("metadata", {}).get("language_info", {}).get("name", "python") ) cells = notebook.get("cells", []) processed_cells = [] for i, cell in enumerate(cells): cell_type = cell.get("cell_type", "code") # Skip if not code or markdown if cell_type not in ["code", "markdown"]: continue # Get source source = cell.get("source", "") if isinstance(source, list): source = "".join(source) # Get execution count for code cells execution_count = None if cell_type == "code": execution_count = cell.get("execution_count") # Process outputs for code cells outputs = [] if cell_type == "code" and "outputs" in cell: for output in cell["outputs"]: output_type = output.get("output_type", "") # Process different output types if output_type == "stream": text = output.get("text", "") if isinstance(text, list): text = "".join(text) outputs.append( NotebookCellOutput(output_type="stream", text=text) ) elif output_type in ["execute_result", "display_data"]: # Process text output text = None if "data" in output and "text/plain" in output["data"]: text_data = output["data"]["text/plain"] if isinstance(text_data, list): text = "".join(text_data) else: text = text_data # Process image output image = None if "data" in output: if "image/png" in output["data"]: image = NotebookOutputImage( image_data=output["data"]["image/png"], media_type="image/png", ) elif "image/jpeg" in output["data"]: image = NotebookOutputImage( image_data=output["data"]["image/jpeg"], media_type="image/jpeg", ) outputs.append( NotebookCellOutput( output_type=output_type, text=text, image=image ) ) elif output_type == "error": # Format error traceback ename = output.get("ename", "") evalue = output.get("evalue", "") traceback = output.get("traceback", []) # Handle raw text strings and lists of strings if isinstance(traceback, list): # Clean ANSI escape codes and join the list but preserve the formatting clean_traceback = [ clean_ansi_escapes(line) for line in traceback ] traceback_text = "\n".join(clean_traceback) else: traceback_text = clean_ansi_escapes(str(traceback)) error_text = f"{ename}: {evalue}\n{traceback_text}" outputs.append( NotebookCellOutput(output_type="error", text=error_text) ) # Create cell object processed_cell = NotebookCellSource( cell_index=i, cell_type=cell_type, source=source, language=language, execution_count=execution_count, outputs=outputs, ) processed_cells.append(processed_cell) return notebook, processed_cells