"""MCP Server implementation."""
import asyncio
import json
from typing import Any, Dict
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from .config import get_config
from .logger import get_logger
from .tools import get_all_tools, get_tool_handler
# Initialize config and logger
config = get_config()
logger = get_logger(__name__)
# Create MCP server instance
server = Server(config.server_name)
@server.list_tools()
async def list_tools() -> list[Tool]:
"""
List all available tools.
Returns:
List of Tool objects
"""
logger.info("list_tools called")
tools = get_all_tools()
logger.debug(f"Returning {len(tools)} tools")
# Convert to MCP Tool format
mcp_tools = [
Tool(
name=tool["name"],
description=tool["description"],
inputSchema=tool["inputSchema"],
)
for tool in tools
]
return mcp_tools
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> list[TextContent]:
"""
Call a tool by name with given arguments.
Args:
name: Tool name
arguments: Tool arguments
Returns:
List of TextContent with results or error
"""
logger.info(f"call_tool: name={name}, arguments={json.dumps(arguments)}")
try:
# Get tool handler
handler = get_tool_handler(name)
# Call the tool (all tools are async)
result = await handler(**arguments)
# Convert result to JSON string
result_str = json.dumps(result, indent=2)
logger.debug(f"Tool {name} succeeded: {result_str}")
return [TextContent(type="text", text=result_str)]
except KeyError as e:
error_msg = f"Tool not found: {name}"
logger.error(error_msg)
return [TextContent(type="text", text=json.dumps({"error": error_msg}))]
except TypeError as e:
error_msg = f"Invalid arguments for tool {name}: {str(e)}"
logger.error(error_msg)
return [TextContent(type="text", text=json.dumps({"error": error_msg}))]
except ValueError as e:
error_msg = f"Validation error in tool {name}: {str(e)}"
logger.warning(error_msg)
return [TextContent(type="text", text=json.dumps({"error": error_msg}))]
except FileNotFoundError as e:
error_msg = f"File not found in tool {name}: {str(e)}"
logger.warning(error_msg)
return [TextContent(type="text", text=json.dumps({"error": error_msg}))]
except FileExistsError as e:
error_msg = f"File already exists in tool {name}: {str(e)}"
logger.warning(error_msg)
return [TextContent(type="text", text=json.dumps({"error": error_msg}))]
except Exception as e:
error_msg = f"Internal error in tool {name}: {str(e)}"
logger.error(error_msg, exc_info=True)
return [TextContent(type="text", text=json.dumps({"error": error_msg}))]
async def main():
"""Run the MCP server."""
logger.info(f"Starting {config.server_name} v{config.server_version}")
logger.info(f"Debug mode: {config.is_debug}")
logger.info(f"Workspace root: {config.workspace_root}")
# Import tools to trigger registration
from . import tools # noqa: F401
logger.info(f"Registered {len(get_all_tools())} tools")
# Run server with stdio transport
async with stdio_server() as (read_stream, write_stream):
logger.info("Server started, listening on stdio...")
await server.run(
read_stream,
write_stream,
server.create_initialization_options(),
)
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info("Server stopped by user")
except Exception as e:
logger.critical(f"Server crashed: {e}", exc_info=True)
raise
