#!/usr/bin/env python3
"""
Excelワークブックプロセッサ - MCPサーバー
Excelワークブックを処理するためのModel Context Protocolサーバー実装
"""
import asyncio
import json
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from .file_handler import ExcelFileResolver, WorkbookParser
# MCPサーバーインスタンスを初期化
mcp_server = Server("excel-workbook-processor")
@mcp_server.list_tools()
async def get_available_operations() -> list[Tool]:
"""
利用可能なすべてのMCPツールを登録して返します。
Returns:
MCPサーバー用のTool定義のリスト
"""
return [
Tool(
name="load_workbook",
description=(
"Loads and extracts data from all sheets in an Excel workbook. "
"Returns a JSON object with sheet names as keys and row arrays as values."
),
inputSchema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": (
"Path to the Excel file. Can be relative to /app/sample or /app/data, "
"or an absolute path."
)
}
},
"required": ["file_path"]
}
),
Tool(
name="get_sheet_by_name",
description=(
"Extracts data from a specific sheet identified by its name. "
"If no sheet name is provided, extracts from the first sheet."
),
inputSchema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": (
"Path to the Excel file. Can be relative to /app/sample or /app/data, "
"or an absolute path."
)
},
"sheet_name": {
"type": "string",
"description": "Name of the target sheet (optional, defaults to first sheet)"
}
},
"required": ["file_path"]
}
),
Tool(
name="get_sheet_by_position",
description=(
"Extracts data from a specific sheet identified by its zero-based index. "
"If no position is provided, extracts from the first sheet (position 0)."
),
inputSchema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": (
"Path to the Excel file. Can be relative to /app/sample or /app/data, "
"or an absolute path."
)
},
"sheet_position": {
"type": "integer",
"description": "Zero-based index of the target sheet (optional, defaults to 0)"
}
},
"required": ["file_path"]
}
)
]
@mcp_server.call_tool()
async def execute_operation(
name: str,
arguments: dict[str, Any]
) -> list[TextContent]:
"""
リクエストされたMCPツール操作を実行します。
Args:
name: 実行するツールの名前
arguments: ツール用の引数の辞書
Returns:
操作結果を含むTextContentオブジェクトのリスト
"""
try:
file_path = arguments.get("file_path")
if not file_path:
raise ValueError("file_path parameter is required")
# ファイルパスを解決
resolved_path = ExcelFileResolver.locate_file(file_path)
# 適切な操作を実行
if name == "load_workbook":
workbook_data = WorkbookParser.extract_all_sheets(resolved_path)
elif name == "get_sheet_by_name":
sheet_name = arguments.get("sheet_name")
if sheet_name:
workbook_data = WorkbookParser.extract_sheet_by_name(
resolved_path,
sheet_name
)
else:
# シート名が指定されていない場合は最初のシートを使用
workbook_data = WorkbookParser.extract_sheet_by_position(
resolved_path,
0
)
elif name == "get_sheet_by_position":
sheet_position = arguments.get("sheet_position", 0)
workbook_data = WorkbookParser.extract_sheet_by_position(
resolved_path,
sheet_position
)
else:
raise ValueError(f"Unknown operation: {name}")
# レスポンスをJSON形式でフォーマット
json_output = json.dumps(
workbook_data,
indent=2,
ensure_ascii=False
)
return [TextContent(
type="text",
text=json_output
)]
except FileNotFoundError as e:
error_msg = f"File not found: {str(e)}"
return [TextContent(type="text", text=error_msg)]
except ValueError as e:
error_msg = f"Invalid input: {str(e)}"
return [TextContent(type="text", text=error_msg)]
except IndexError as e:
error_msg = f"Index error: {str(e)}"
return [TextContent(type="text", text=error_msg)]
except Exception as e:
error_msg = f"Unexpected error: {str(e)}"
return [TextContent(type="text", text=error_msg)]
async def start_server():
"""
MCPサーバーを起動し、stdio経由でリクエストの処理を開始します。
"""
async with stdio_server() as (read_stream, write_stream):
await mcp_server.run(
read_stream,
write_stream,
mcp_server.create_initialization_options()
)
def main():
"""MCPサーバーアプリケーションのエントリーポイント"""
asyncio.run(start_server())
if __name__ == "__main__":
main()
