from mcp.server.fastmcp import FastMCP, Context
import socket
import json
import asyncio
import logging
from dataclasses import dataclass
from contextlib import asynccontextmanager
from typing import AsyncIterator, Dict, Any, List
# Configure logging
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger("SketchupMCPServer")
# Define version directly to avoid pkg_resources dependency
__version__ = "0.1.17"
logger.info(f"SketchupMCP Server version {__version__} starting up")
@dataclass
class SketchupConnection:
host: str
port: int
sock: socket.socket = None
def connect(self) -> bool:
"""Connect to the Sketchup extension socket server"""
if self.sock:
try:
# Test if connection is still alive
self.sock.settimeout(0.1)
self.sock.send(b'')
return True
except (socket.error, BrokenPipeError, ConnectionResetError):
# Connection is dead, close it and reconnect
logger.info("Connection test failed, reconnecting...")
self.disconnect()
try:
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.connect((self.host, self.port))
logger.info(f"Connected to Sketchup at {self.host}:{self.port}")
return True
except Exception as e:
logger.error(f"Failed to connect to Sketchup: {str(e)}")
self.sock = None
return False
def disconnect(self):
"""Disconnect from the Sketchup extension"""
if self.sock:
try:
self.sock.close()
except Exception as e:
logger.error(f"Error disconnecting from Sketchup: {str(e)}")
finally:
self.sock = None
def receive_full_response(self, sock, buffer_size=8192):
"""Receive the complete response, potentially in multiple chunks"""
chunks = []
sock.settimeout(15.0)
try:
while True:
try:
chunk = sock.recv(buffer_size)
if not chunk:
if not chunks:
raise Exception("Connection closed before receiving any data")
break
chunks.append(chunk)
try:
data = b''.join(chunks)
json.loads(data.decode('utf-8'))
logger.info(f"Received complete response ({len(data)} bytes)")
return data
except json.JSONDecodeError:
continue
except socket.timeout:
logger.warning("Socket timeout during chunked receive")
break
except (ConnectionError, BrokenPipeError, ConnectionResetError) as e:
logger.error(f"Socket connection error during receive: {str(e)}")
raise
except socket.timeout:
logger.warning("Socket timeout during chunked receive")
except Exception as e:
logger.error(f"Error during receive: {str(e)}")
raise
if chunks:
data = b''.join(chunks)
logger.info(f"Returning data after receive completion ({len(data)} bytes)")
try:
json.loads(data.decode('utf-8'))
return data
except json.JSONDecodeError:
raise Exception("Incomplete JSON response received")
else:
raise Exception("No data received")
def send_command(self, method: str, params: Dict[str, Any] = None, request_id: Any = None) -> Dict[str, Any]:
"""Send a JSON-RPC request to Sketchup and return the response"""
# Try to connect if not connected
if not self.connect():
raise ConnectionError("Not connected to Sketchup")
# Ensure we're sending a proper JSON-RPC request
if method == "tools/call" and params and "name" in params and "arguments" in params:
# This is already in the correct format
request = {
"jsonrpc": "2.0",
"method": method,
"params": params,
"id": request_id
}
else:
# This is a direct command - convert to JSON-RPC
command_name = method
command_params = params or {}
# Log the conversion
logger.info(f"Converting direct command '{command_name}' to JSON-RPC format")
request = {
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": command_name,
"arguments": command_params
},
"id": request_id
}
# Maximum number of retries
max_retries = 2
retry_count = 0
while retry_count <= max_retries:
try:
logger.info(f"Sending JSON-RPC request: {request}")
# Log the exact bytes being sent
request_bytes = json.dumps(request).encode('utf-8') + b'\n'
logger.info(f"Raw bytes being sent: {request_bytes}")
self.sock.sendall(request_bytes)
logger.info(f"Request sent, waiting for response...")
self.sock.settimeout(15.0)
response_data = self.receive_full_response(self.sock)
logger.info(f"Received {len(response_data)} bytes of data")
response = json.loads(response_data.decode('utf-8'))
logger.info(f"Response parsed: {response}")
if "error" in response:
logger.error(f"Sketchup error: {response['error']}")
raise Exception(response["error"].get("message", "Unknown error from Sketchup"))
return response.get("result", {})
except (socket.timeout, ConnectionError, BrokenPipeError, ConnectionResetError) as e:
logger.warning(f"Connection error (attempt {retry_count+1}/{max_retries+1}): {str(e)}")
retry_count += 1
if retry_count <= max_retries:
logger.info(f"Retrying connection...")
self.disconnect()
if not self.connect():
logger.error("Failed to reconnect")
break
else:
logger.error(f"Max retries reached, giving up")
self.sock = None
raise Exception(f"Connection to Sketchup lost after {max_retries+1} attempts: {str(e)}")
except json.JSONDecodeError as e:
logger.error(f"Invalid JSON response from Sketchup: {str(e)}")
if 'response_data' in locals() and response_data:
logger.error(f"Raw response (first 200 bytes): {response_data[:200]}")
raise Exception(f"Invalid response from Sketchup: {str(e)}")
except Exception as e:
logger.error(f"Error communicating with Sketchup: {str(e)}")
self.sock = None
raise Exception(f"Communication error with Sketchup: {str(e)}")
# Global connection management
_sketchup_connection = None
def get_sketchup_connection():
"""Get or create a persistent Sketchup connection"""
global _sketchup_connection
if _sketchup_connection is not None:
try:
# Test connection with a ping command
ping_request = {
"jsonrpc": "2.0",
"method": "ping",
"params": {},
"id": 0
}
_sketchup_connection.sock.sendall(json.dumps(ping_request).encode('utf-8') + b'\n')
return _sketchup_connection
except Exception as e:
logger.warning(f"Existing connection is no longer valid: {str(e)}")
try:
_sketchup_connection.disconnect()
except:
pass
_sketchup_connection = None
if _sketchup_connection is None:
_sketchup_connection = SketchupConnection(host="localhost", port=9876)
if not _sketchup_connection.connect():
logger.error("Failed to connect to Sketchup")
_sketchup_connection = None
raise Exception("Could not connect to Sketchup. Make sure the Sketchup extension is running.")
logger.info("Created new persistent connection to Sketchup")
return _sketchup_connection
@asynccontextmanager
async def server_lifespan(server: FastMCP) -> AsyncIterator[Dict[str, Any]]:
"""Manage server startup and shutdown lifecycle"""
try:
logger.info("SketchupMCP server starting up")
try:
sketchup = get_sketchup_connection()
logger.info("Successfully connected to Sketchup on startup")
except Exception as e:
logger.warning(f"Could not connect to Sketchup on startup: {str(e)}")
logger.warning("Make sure the Sketchup extension is running")
yield {}
finally:
global _sketchup_connection
if _sketchup_connection:
logger.info("Disconnecting from Sketchup")
_sketchup_connection.disconnect()
_sketchup_connection = None
logger.info("SketchupMCP server shut down")
# Create MCP server with lifespan support
mcp = FastMCP(
"SketchupMCP",
description="Sketchup integration through the Model Context Protocol",
lifespan=server_lifespan
)
# Tool endpoints
@mcp.tool()
def create_component(
ctx: Context,
type: str = "cube",
position: List[float] = None,
dimensions: List[float] = None
) -> str:
"""Create a new component in Sketchup"""
try:
logger.info(f"create_component called with type={type}, position={position}, dimensions={dimensions}, request_id={ctx.request_id}")
sketchup = get_sketchup_connection()
params = {
"name": "create_component",
"arguments": {
"type": type,
"position": position or [0,0,0],
"dimensions": dimensions or [1,1,1]
}
}
logger.info(f"Calling send_command with method='tools/call', params={params}, request_id={ctx.request_id}")
result = sketchup.send_command(
method="tools/call",
params=params,
request_id=ctx.request_id
)
logger.info(f"create_component result: {result}")
return json.dumps(result)
except Exception as e:
logger.error(f"Error in create_component: {str(e)}")
return f"Error creating component: {str(e)}"
@mcp.tool()
def delete_component(
ctx: Context,
id: str
) -> str:
"""Delete a component by ID"""
try:
sketchup = get_sketchup_connection()
result = sketchup.send_command(
method="tools/call",
params={
"name": "delete_component",
"arguments": {"id": id}
},
request_id=ctx.request_id
)
return json.dumps(result)
except Exception as e:
return f"Error deleting component: {str(e)}"
@mcp.tool()
def transform_component(
ctx: Context,
id: str,
position: List[float] = None,
rotation: List[float] = None,
scale: List[float] = None
) -> str:
"""Transform a component's position, rotation, or scale"""
try:
sketchup = get_sketchup_connection()
arguments = {"id": id}
if position is not None:
arguments["position"] = position
if rotation is not None:
arguments["rotation"] = rotation
if scale is not None:
arguments["scale"] = scale
result = sketchup.send_command(
method="tools/call",
params={
"name": "transform_component",
"arguments": arguments
},
request_id=ctx.request_id
)
return json.dumps(result)
except Exception as e:
return f"Error transforming component: {str(e)}"
@mcp.tool()
def get_selection(ctx: Context) -> str:
"""Get currently selected components"""
try:
sketchup = get_sketchup_connection()
result = sketchup.send_command(
method="tools/call",
params={
"name": "get_selection",
"arguments": {}
},
request_id=ctx.request_id
)
return json.dumps(result)
except Exception as e:
return f"Error getting selection: {str(e)}"
@mcp.tool()
def set_material(
ctx: Context,
id: str,
material: str
) -> str:
"""Set material for a component"""
try:
sketchup = get_sketchup_connection()
result = sketchup.send_command(
method="tools/call",
params={
"name": "set_material",
"arguments": {
"id": id,
"material": material
}
},
request_id=ctx.request_id
)
return json.dumps(result)
except Exception as e:
return f"Error setting material: {str(e)}"
@mcp.tool()
def export_scene(
ctx: Context,
format: str = "skp"
) -> str:
"""Export the current scene"""
try:
sketchup = get_sketchup_connection()
result = sketchup.send_command(
method="tools/call",
params={
"name": "export",
"arguments": {
"format": format
}
},
request_id=ctx.request_id
)
return json.dumps(result)
except Exception as e:
return f"Error exporting scene: {str(e)}"
@mcp.tool()
def create_mortise_tenon(
ctx: Context,
mortise_id: str,
tenon_id: str,
width: float = 1.0,
height: float = 1.0,
depth: float = 1.0,
offset_x: float = 0.0,
offset_y: float = 0.0,
offset_z: float = 0.0
) -> str:
"""Create a mortise and tenon joint between two components"""
try:
logger.info(f"create_mortise_tenon called with mortise_id={mortise_id}, tenon_id={tenon_id}, width={width}, height={height}, depth={depth}, offsets=({offset_x}, {offset_y}, {offset_z})")
sketchup = get_sketchup_connection()
result = sketchup.send_command(
method="tools/call",
params={
"name": "create_mortise_tenon",
"arguments": {
"mortise_id": mortise_id,
"tenon_id": tenon_id,
"width": width,
"height": height,
"depth": depth,
"offset_x": offset_x,
"offset_y": offset_y,
"offset_z": offset_z
}
},
request_id=ctx.request_id
)
logger.info(f"create_mortise_tenon result: {result}")
return json.dumps(result)
except Exception as e:
logger.error(f"Error in create_mortise_tenon: {str(e)}")
return f"Error creating mortise and tenon joint: {str(e)}"
@mcp.tool()
def create_dovetail(
ctx: Context,
tail_id: str,
pin_id: str,
width: float = 1.0,
height: float = 1.0,
depth: float = 1.0,
angle: float = 15.0,
num_tails: int = 3,
offset_x: float = 0.0,
offset_y: float = 0.0,
offset_z: float = 0.0
) -> str:
"""Create a dovetail joint between two components"""
try:
logger.info(f"create_dovetail called with tail_id={tail_id}, pin_id={pin_id}, width={width}, height={height}, depth={depth}, angle={angle}, num_tails={num_tails}")
sketchup = get_sketchup_connection()
result = sketchup.send_command(
method="tools/call",
params={
"name": "create_dovetail",
"arguments": {
"tail_id": tail_id,
"pin_id": pin_id,
"width": width,
"height": height,
"depth": depth,
"angle": angle,
"num_tails": num_tails,
"offset_x": offset_x,
"offset_y": offset_y,
"offset_z": offset_z
}
},
request_id=ctx.request_id
)
logger.info(f"create_dovetail result: {result}")
return json.dumps(result)
except Exception as e:
logger.error(f"Error in create_dovetail: {str(e)}")
return f"Error creating dovetail joint: {str(e)}"
@mcp.tool()
def create_finger_joint(
ctx: Context,
board1_id: str,
board2_id: str,
width: float = 1.0,
height: float = 1.0,
depth: float = 1.0,
num_fingers: int = 5,
offset_x: float = 0.0,
offset_y: float = 0.0,
offset_z: float = 0.0
) -> str:
"""Create a finger joint (box joint) between two components"""
try:
logger.info(f"create_finger_joint called with board1_id={board1_id}, board2_id={board2_id}, width={width}, height={height}, depth={depth}, num_fingers={num_fingers}")
sketchup = get_sketchup_connection()
result = sketchup.send_command(
method="tools/call",
params={
"name": "create_finger_joint",
"arguments": {
"board1_id": board1_id,
"board2_id": board2_id,
"width": width,
"height": height,
"depth": depth,
"num_fingers": num_fingers,
"offset_x": offset_x,
"offset_y": offset_y,
"offset_z": offset_z
}
},
request_id=ctx.request_id
)
logger.info(f"create_finger_joint result: {result}")
return json.dumps(result)
except Exception as e:
logger.error(f"Error in create_finger_joint: {str(e)}")
return f"Error creating finger joint: {str(e)}"
@mcp.tool()
def eval_ruby(
ctx: Context,
code: str
) -> str:
"""Evaluate arbitrary Ruby code in Sketchup"""
try:
logger.info(f"eval_ruby called with code length: {len(code)}")
sketchup = get_sketchup_connection()
result = sketchup.send_command(
method="tools/call",
params={
"name": "eval_ruby",
"arguments": {
"code": code
}
},
request_id=ctx.request_id
)
logger.info(f"eval_ruby result: {result}")
# Format the response to include the result
response = {
"success": True,
"result": result.get("content", [{"text": "Success"}])[0].get("text", "Success") if isinstance(result.get("content"), list) and len(result.get("content", [])) > 0 else "Success"
}
return json.dumps(response)
except Exception as e:
logger.error(f"Error in eval_ruby: {str(e)}")
return json.dumps({
"success": False,
"error": str(e)
})
def main():
mcp.run()
if __name__ == "__main__":
main()