Excel Reader Server

  • src
  • excel_reader_server
#!/usr/bin/env python3 import openpyxl from mcp.server.models import InitializationOptions import mcp.types as types from mcp.server import NotificationOptions, Server import mcp.server.stdio import os import json server = Server("excel-reader-server") @server.list_tools() async def handle_list_tools() -> list[types.Tool]: """ List available tools. Each tool specifies its arguments using JSON Schema validation. """ return [ types.Tool( name="read_excel", description="Read content from Excel (xlsx) files", inputSchema={ "type": "object", "properties": { "file_path": { "type": "string", "description": "Path to the Excel file" } }, "required": ["file_path"] } ), types.Tool( name="read_excel_by_sheet_name", description="Read content from a specific sheet by name in Excel (xlsx) files. Reads first sheet if sheet_name not provided.", inputSchema={ "type": "object", "properties": { "file_path": { "type": "string", "description": "Path to the Excel file" }, "sheet_name": { "type": "string", "description": "Name of the sheet to read (optional, defaults to first sheet)" } }, "required": ["file_path"] } ), types.Tool( name="read_excel_by_sheet_index", description="Read content from a specific sheet by index in Excel (xlsx) files. Reads first sheet (index 0) if sheet_index not provided.", inputSchema={ "type": "object", "properties": { "file_path": { "type": "string", "description": "Path to the Excel file" }, "sheet_index": { "type": "integer", "description": "Index of the sheet to read (optional, defaults to 0)", "minimum": 0 } }, "required": ["file_path"] } ) ] @server.call_tool() async def handle_call_tool( name: str, arguments: dict | None ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: """ Handle tool execution requests. """ if name not in ["read_excel", "read_excel_by_sheet_name", "read_excel_by_sheet_index"]: raise ValueError(f"Unknown tool: {name}") if not arguments: raise ValueError("Missing arguments") file_path = arguments.get("file_path") if not file_path: raise ValueError("file_path is required") abs_path = os.path.abspath(file_path) if not os.path.exists(abs_path): raise ValueError(f"Excel file not found: {file_path}") try: # Load the workbook workbook = openpyxl.load_workbook(abs_path, data_only=True) result = {} if name == "read_excel": # Original functionality - process all sheets for sheet_name in workbook.sheetnames: sheet = workbook[sheet_name] sheet_data = [] for row in sheet.rows: row_data = [str(cell.value) if cell.value is not None else "" for cell in row] sheet_data.append(row_data) result[sheet_name] = sheet_data elif name == "read_excel_by_sheet_name": # Get sheet by name, default to first sheet if not specified sheet_name = arguments.get("sheet_name") if not sheet_name: sheet_name = workbook.sheetnames[0] elif sheet_name not in workbook.sheetnames: raise ValueError(f"Sheet '{sheet_name}' not found in workbook") sheet = workbook[sheet_name] sheet_data = [] for row in sheet.rows: row_data = [str(cell.value) if cell.value is not None else "" for cell in row] sheet_data.append(row_data) result[sheet_name] = sheet_data elif name == "read_excel_by_sheet_index": # Get sheet by index, default to 0 if not specified sheet_index = arguments.get("sheet_index", 0) if sheet_index < 0 or sheet_index >= len(workbook.sheetnames): raise ValueError(f"Sheet index {sheet_index} is out of range") sheet_name = workbook.sheetnames[sheet_index] sheet = workbook[sheet_name] sheet_data = [] for row in sheet.rows: row_data = [str(cell.value) if cell.value is not None else "" for cell in row] sheet_data.append(row_data) result[sheet_name] = sheet_data return [ types.TextContent( type="text", text=json.dumps(result, indent=2) ) ] except Exception as e: raise ValueError(f"Error reading Excel file: {str(e)}") async def main(): # Run the server using stdin/stdout streams async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_name="excel-reader-server", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), )