Feat/add stdio transport (#25)

* feat: Add stdio transport method.
Update relevant documentation and code to support both SSE and stdio connection methods, initialize the EXCEL_FILES_PATH variable and perform the corresponding path handling.

* Remove PyPI mirror configuration.
This commit is contained in:
allenZhang
2025-05-17 22:39:06 +08:00
committed by GitHub
parent 46adf0b173
commit bbab5bf184
4 changed files with 131 additions and 54 deletions

View File

@ -1,13 +1,19 @@
import asyncio
from .server import run_server
import typer
from typing import Optional
def main():
"""Start the Excel MCP server."""
from .server import run_sse, run_stdio
app = typer.Typer(help="Excel MCP Server")
@app.command()
def sse():
"""Start Excel MCP Server in SSE mode"""
print("Excel MCP Server - SSE mode")
print("----------------------")
print("Press Ctrl+C to exit")
try:
print("Excel MCP Server")
print("---------------")
print("Starting server... Press Ctrl+C to exit")
asyncio.run(run_server())
asyncio.run(run_sse())
except KeyboardInterrupt:
print("\nShutting down server...")
except Exception as e:
@ -15,7 +21,21 @@ def main():
import traceback
traceback.print_exc()
finally:
print("Server stopped.")
print("Service stopped.")
@app.command()
def stdio():
"""Start Excel MCP Server in stdio mode"""
try:
run_stdio()
except KeyboardInterrupt:
print("\nShutting down server...")
except Exception as e:
print(f"\nError: {e}")
import traceback
traceback.print_exc()
finally:
print("Service stopped.")
if __name__ == "__main__":
main()
app()

View File

@ -34,25 +34,31 @@ from excel_mcp.sheet import (
unmerge_range,
)
# Get project root directory path for log file path.
# When using the stdio transmission method,
# relative paths may cause log files to fail to create
# due to the client's running location and permission issues,
# resulting in the program not being able to run.
# Thus using os.path.join(ROOT_DIR, "excel-mcp.log") instead.
ROOT_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
LOG_FILE = os.path.join(ROOT_DIR, "excel-mcp.log")
# Initialize EXCEL_FILES_PATH variable without assigning a value
EXCEL_FILES_PATH = None
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[
logging.StreamHandler(sys.stdout),
logging.FileHandler("excel-mcp.log")
# Referring to https://github.com/modelcontextprotocol/python-sdk/issues/409#issuecomment-2816831318
# The stdio mode server MUST NOT write anything to its stdout that is not a valid MCP message.
logging.FileHandler(LOG_FILE)
],
force=True
)
logger = logging.getLogger("excel-mcp")
# Get Excel files path from environment or use default
EXCEL_FILES_PATH = os.environ.get("EXCEL_FILES_PATH", "./excel_files")
# Create the directory if it doesn't exist
os.makedirs(EXCEL_FILES_PATH, exist_ok=True)
# Initialize FastMCP server
mcp = FastMCP(
"excel-mcp",
@ -80,8 +86,13 @@ def get_excel_path(filename: str) -> str:
# If filename is already an absolute path, return it
if os.path.isabs(filename):
return filename
# Use the configured Excel files path
# Check if in SSE mode (EXCEL_FILES_PATH is not None)
if EXCEL_FILES_PATH is None:
# Must use absolute path
raise ValueError(f"Invalid filename: {filename}, must be an absolute path when not in SSE mode")
# In SSE mode, if it's a relative path, resolve it based on EXCEL_FILES_PATH
return os.path.join(EXCEL_FILES_PATH, filename)
@mcp.tool()
@ -491,10 +502,16 @@ def validate_excel_range(
logger.error(f"Error validating range: {e}")
raise
async def run_server():
"""Run the Excel MCP server."""
async def run_sse():
"""Run Excel MCP server in SSE mode."""
# Assign value to EXCEL_FILES_PATH in SSE mode
global EXCEL_FILES_PATH
EXCEL_FILES_PATH = os.environ.get("EXCEL_FILES_PATH", "./excel_files")
# Create directory if it doesn't exist
os.makedirs(EXCEL_FILES_PATH, exist_ok=True)
try:
logger.info(f"Starting Excel MCP server (files directory: {EXCEL_FILES_PATH})")
logger.info(f"Starting Excel MCP server with SSE transport (files directory: {EXCEL_FILES_PATH})")
await mcp.run_sse_async()
except KeyboardInterrupt:
logger.info("Server stopped by user")
@ -503,4 +520,19 @@ async def run_server():
logger.error(f"Server failed: {e}")
raise
finally:
logger.info("Server shutdown complete")
logger.info("Server shutdown complete")
def run_stdio():
"""Run Excel MCP server in stdio mode."""
# No need to assign EXCEL_FILES_PATH in stdio mode
try:
logger.info("Starting Excel MCP server with stdio transport")
mcp.run(transport="stdio")
except KeyboardInterrupt:
logger.info("Server stopped by user")
except Exception as e:
logger.error(f"Server failed: {e}")
raise
finally:
logger.info("Server shutdown complete")