server.py•7.2 kB
#!/usr/bin/env python3
"""
MCP Code Assistant Server
A minimal Model Context Protocol (MCP) server implementation that provides
developer tools for file operations, compilation, and execution.
This server communicates via JSON-RPC 2.0 over stdin/stdout.
"""
import sys
import json
from typing import Any, Dict
from logger import log_info, log_error, log_debug
from jsonrpc import (
parse_request,
create_response,
create_error_response,
JSONRPCError,
INTERNAL_ERROR
)
from dispatcher import ToolDispatcher
from tools import ALL_TOOLS
class MCPServer:
"""Main MCP server implementation."""
def __init__(self):
"""Initialize the MCP server."""
self.dispatcher = ToolDispatcher()
self.initialized = False
self.server_info = {
"name": "mcp-code-assistant",
"version": "0.1.0"
}
# Register all available tools
self._register_tools()
log_info(f"MCP Server initialized: {self.server_info['name']} v{self.server_info['version']}")
def _register_tools(self):
"""Register all tools from the tools package."""
for tool_name, tool_def in ALL_TOOLS.items():
self.dispatcher.register_tool(
name=tool_name,
handler=tool_def["handler"],
description=tool_def["description"],
input_schema=tool_def["inputSchema"]
)
def handle_initialize(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Handle the initialize request.
Args:
params: Initialization parameters from client
Returns:
Server capabilities and info
"""
log_info("Received initialize request")
log_debug(f"Client info: {params.get('clientInfo', {})}")
self.initialized = True
return {
"protocolVersion": "2024-11-05",
"serverInfo": self.server_info,
"capabilities": {
"tools": {}
}
}
def handle_tools_list(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Handle tools/list request.
Args:
params: Request parameters
Returns:
List of available tools
"""
log_debug("Listing available tools")
tools = self.dispatcher.get_tool_list()
return {
"tools": tools
}
def handle_tools_call(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Handle tools/call request.
Args:
params: Must contain 'name' and 'arguments'
Returns:
Tool execution result
Raises:
JSONRPCError: If parameters are invalid or tool fails
"""
if "name" not in params:
raise JSONRPCError(-32602, "Missing 'name' parameter")
if "arguments" not in params:
raise JSONRPCError(-32602, "Missing 'arguments' parameter")
tool_name = params["name"]
arguments = params["arguments"]
log_info(f"Calling tool: {tool_name}")
# Dispatch to tool handler
result = self.dispatcher.dispatch(tool_name, arguments)
return {
"content": [
{
"type": "text",
"text": json.dumps(result, indent=2)
}
]
}
def handle_request(self, request: Dict[str, Any]) -> str:
"""
Handle a single JSON-RPC request.
Args:
request: Parsed JSON-RPC request
Returns:
JSON-RPC response string
"""
method = request.get("method")
params = request.get("params", {})
request_id = request.get("id")
log_debug(f"Handling method: {method}")
try:
# Route to appropriate handler
if method == "initialize":
result = self.handle_initialize(params)
elif method == "tools/list":
result = self.handle_tools_list(params)
elif method == "tools/call":
result = self.handle_tools_call(params)
else:
raise JSONRPCError(-32601, f"Method not found: {method}")
# Create success response
return create_response(request_id, result)
except JSONRPCError as e:
log_error(f"JSON-RPC error: {e.message}")
return create_error_response(request_id, e.code, e.message, e.data)
except Exception as e:
log_error(f"Unexpected error: {e}")
return create_error_response(
request_id,
INTERNAL_ERROR,
"Internal server error",
str(e)
)
def run(self):
"""
Main server loop - read from stdin, process requests, write to stdout.
"""
log_info("Server starting - listening on stdin")
try:
for line in sys.stdin:
line = line.strip()
if not line:
continue
log_debug(f"Received: {line[:100]}...")
try:
# Parse the request
request = parse_request(line)
# Handle the request
response = self.handle_request(request)
# Write response to stdout
print(response, flush=True)
log_debug(f"Sent: {response[:100]}...")
except JSONRPCError as e:
# Send error response
error_response = create_error_response(
None,
e.code,
e.message,
e.data
)
print(error_response, flush=True)
log_error(f"Request error: {e.message}")
except Exception as e:
# Unexpected error during request handling
log_error(f"Fatal error processing request: {e}")
error_response = create_error_response(
None,
INTERNAL_ERROR,
"Internal server error",
str(e)
)
print(error_response, flush=True)
except KeyboardInterrupt:
log_info("Server interrupted by user")
except Exception as e:
log_error(f"Fatal server error: {e}")
sys.exit(1)
finally:
log_info("Server shutting down")
def main():
"""Entry point for the MCP server."""
server = MCPServer()
server.run()
if __name__ == "__main__":
main()