# Directory Structure
```
├── .gitignore
├── ai
│ ├── PROMPTS.md
│ └── rhino-python-bridge-docs.md
├── claude_adapter.py
├── config
│ └── default_config.json
├── docs
│ └── ARCHITECTURE.md
├── README.md
├── requirements-dev.txt
├── requirements.txt
├── rhino-python-bridge-docs.md
├── setup.py
├── src
│ ├── __init__.py
│ ├── rhino_mcp
│ │ ├── __init__.py
│ │ ├── mcp_server.py
│ │ └── rhino_client.py
│ └── rhino_plugin
│ ├── __init__.py
│ └── rhino_server.py
├── test_create_curve.py
├── test_curve.py
├── test_mcp_client.py
├── tests
│ └── rhino_mcp
│ └── test_rhino_client.py
└── ws_adapter.py
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# Virtual environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# IDE files
.idea/
.vscode/
*.swp
*.swo
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
# Jupyter Notebook
.ipynb_checkpoints
# Logs
*.log
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# RhinoMCP
RhinoMCP connects Rhino3D to Claude AI via the Model Context Protocol (MCP), enabling Claude to directly interact with and control Rhino3D for AI-assisted 3D modeling, analysis, and design workflows.
## Project Overview
This integration consists of two main components:
1. **Rhino Plugin**: A socket server that runs inside Rhino's Python editor, providing a communication interface to Rhino's functionality.
2. **MCP Server**: An implementation of the Model Context Protocol that connects Claude AI to the Rhino plugin, enabling AI-controlled operations.
## Features
- Socket-based bidirectional communication between Python and Rhino
- Model Context Protocol server for Claude AI integration
- Support for NURBS curve creation (initial test feature)
- Python script execution within Rhino's context
- Compatible with both Claude Desktop and Windsurf as clients
## Installation
### Requirements
- Rhinoceros 3D (Version 7 or 8)
- Python 3.10 or higher
- Windows 10 or 11
### Install Using uv (Recommended)
```bash
# Create and activate a virtual environment
mkdir -p .venv
uv venv .venv
source .venv/Scripts/activate # On Windows with Git Bash
# Install the package
uv pip install -e .
```
### Install Using pip
```bash
# Create and activate a virtual environment
python -m venv .venv
.venv\Scripts\activate # On Windows
# Install the package
pip install -e .
```
## Usage
### Step 1: Start the Rhino Bridge Server
1. Open Rhino
2. Type `EditPythonScript` in the command line to open Rhino's Python editor
3. Open the Rhino server script from `src/rhino_plugin/rhino_server.py`
4. Run the script (F5 or click the Run button)
5. Verify you see "Rhino Bridge started!" in the output panel
### Step 2: Start the MCP Server
```bash
# Activate your virtual environment
source .venv/Scripts/activate # On Windows with Git Bash
# Start the MCP server
rhinomcp
```
Or run with custom settings:
```bash
rhinomcp --host 127.0.0.1 --port 5000 --rhino-host 127.0.0.1 --rhino-port 8888 --debug
```
### Step 3: Connect with Claude Desktop or Windsurf
Configure Claude Desktop or Windsurf to connect to the MCP server at:
```
ws://127.0.0.1:5000
```
### Example: Creating a NURBS Curve
When connected to Claude, you can ask it to create a NURBS curve in Rhino with a prompt like:
```
Create a NURBS curve in Rhino using points at (0,0,0), (5,10,0), (10,0,0), and (15,10,0).
```
## Development
### Setup Development Environment
```bash
# Clone the repository
git clone https://github.com/FernandoMaytorena/RhinoMCP.git
cd RhinoMCP
# Create and activate virtual environment
uv venv .venv
source .venv/Scripts/activate # On Windows with Git Bash
# Install development dependencies
uv pip install -e ".[dev]"
```
### Run Tests
```bash
pytest
```
### Code Style
This project uses Ruff for linting and formatting:
```bash
ruff check .
ruff format .
```
## Project Structure
```
RhinoMCP/
├── src/
│ ├── rhino_plugin/ # Code that runs inside Rhino
│ │ └── rhino_server.py
│ └── rhino_mcp/ # MCP server implementation
│ ├── rhino_client.py
│ └── mcp_server.py
├── tests/ # Test modules
├── docs/ # Documentation
├── config/ # Configuration files
├── ai/ # AI documentation and prompts
├── setup.py # Package installation
├── requirements.txt # Package dependencies
└── README.md # Project documentation
```
## License
[MIT License](LICENSE)
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
websockets>=11.0.0
typing-extensions>=4.0.0
```
--------------------------------------------------------------------------------
/requirements-dev.txt:
--------------------------------------------------------------------------------
```
-r requirements.txt
pytest>=7.0.0
pytest-mock>=3.10.0
pytest-asyncio>=0.21.0
ruff>=0.0.270
mypy>=1.0.0
```
--------------------------------------------------------------------------------
/src/__init__.py:
--------------------------------------------------------------------------------
```python
"""RhinoMCP package initialization.
This package connects Rhino3D to Claude AI via the Model Context Protocol.
"""
__version__ = "0.1.0"
```
--------------------------------------------------------------------------------
/src/rhino_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
"""Rhino MCP module for RhinoMCP.
This module implements the Model Context Protocol server that connects to Claude AI
and communicates with the Rhino plugin via socket connection.
"""
```
--------------------------------------------------------------------------------
/src/rhino_plugin/__init__.py:
--------------------------------------------------------------------------------
```python
"""Rhino Plugin module for RhinoMCP.
This module contains the components that run inside Rhino's Python environment
and establishes a socket server to communicate with external Python clients.
"""
```
--------------------------------------------------------------------------------
/config/default_config.json:
--------------------------------------------------------------------------------
```json
{
"rhino_server": {
"host": "127.0.0.1",
"port": 8888
},
"mcp_server": {
"host": "127.0.0.1",
"port": 5000
},
"logging": {
"level": "INFO",
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
"file": null
}
}
```
--------------------------------------------------------------------------------
/test_curve.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""
Test script to create a NURBS curve in Rhino via RhinoMCP.
"""
from src.rhino_mcp.rhino_client import RhinoClient, Point3d
def main():
# Create a client and connect to Rhino
client = RhinoClient()
if client.connect():
print("Connected to Rhino Bridge")
# Create the points as dictionaries (Point3d is a TypedDict)
points = [
{"x": 0, "y": 0, "z": 0},
{"x": 10, "y": 10, "z": 0},
{"x": 20, "y": 0, "z": 0}
]
# Send the curve creation command
result = client.create_curve(points)
# Print the result
print("Curve creation result:", result)
# Disconnect
client.disconnect()
else:
print("Failed to connect to Rhino Bridge")
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/setup.py:
--------------------------------------------------------------------------------
```python
"""RhinoMCP package setup script."""
from setuptools import setup, find_packages
setup(
name="rhinomcp",
version="0.1.0",
description="Connect Rhino3D to Claude AI via the Model Context Protocol",
author="Fernando Maytorena",
author_email="",
url="https://github.com/FernandoMaytorena/RhinoMCP",
package_dir={"": "src"},
packages=find_packages(where="src"),
python_requires=">=3.10",
install_requires=[
"websockets>=11.0.0",
"typing-extensions>=4.0.0",
],
extras_require={
"dev": [
"pytest>=7.0.0",
"pytest-mock>=3.10.0",
"pytest-asyncio>=0.21.0",
"ruff>=0.0.270",
"mypy>=1.0.0",
]
},
entry_points={
"console_scripts": [
"rhinomcp=rhino_mcp.mcp_server:main",
],
},
classifiers=[
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10",
],
)
```
--------------------------------------------------------------------------------
/test_create_curve.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""
Test script to create a curve in Rhino using the RhinoClient.
This script connects directly to the Rhino Bridge and creates a NURBS curve
with three points.
"""
import sys
import json
from rhino_mcp.rhino_client import RhinoClient
def main():
"""Connect to Rhino and create a curve."""
# Create a client and connect to Rhino
client = RhinoClient(host="127.0.0.1", port=8888)
connected = client.connect()
if not connected:
print("Failed to connect to Rhino Bridge")
return 1
print("Connected to Rhino Bridge")
# Create points for the curve
points = [
{"x": 0, "y": 0, "z": 0},
{"x": 10, "y": 10, "z": 0},
{"x": 20, "y": 0, "z": 0}
]
# Format the command as expected by the Rhino Bridge server
command = {
"type": "create_curve",
"data": {
"points": points
}
}
# Send the command to Rhino
try:
# The client.send_command method adds the type field, so we need to extract our data
response = client.send_command("create_curve", {"points": points})
print(f"Response from Rhino: {response}")
return 0
except Exception as e:
print(f"Error creating curve: {e}")
return 1
finally:
# Clean up
client.disconnect()
if __name__ == "__main__":
sys.exit(main())
```
--------------------------------------------------------------------------------
/ws_adapter.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""
WebSocket adapter for connecting Windsurf to the RhinoMCP server.
This script forwards messages between Windsurf and the RhinoMCP server,
acting as an adapter layer that handles the WebSocket connection.
"""
import asyncio
import json
import sys
import websockets
from typing import Dict, Any, Optional, List
import logging
from websockets.exceptions import ConnectionClosed
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler("ws_adapter.log"),
logging.StreamHandler(sys.stderr)
]
)
logger = logging.getLogger("ws-adapter")
# Target RhinoMCP server URL
TARGET_URL = "ws://127.0.0.1:5000"
async def forward_messages():
"""Forward messages between stdin/stdout and the WebSocket server."""
try:
logger.info(f"Connecting to RhinoMCP server at {TARGET_URL}")
async with websockets.connect(TARGET_URL) as websocket:
logger.info("Connected to RhinoMCP server")
# Start tasks for handling stdin->websocket and websocket->stdout
stdin_task = asyncio.create_task(forward_stdin_to_websocket(websocket))
ws_task = asyncio.create_task(forward_websocket_to_stdout(websocket))
# Wait for either task to complete (or fail)
done, pending = await asyncio.wait(
[stdin_task, ws_task],
return_when=asyncio.FIRST_COMPLETED
)
# Cancel any pending tasks
for task in pending:
task.cancel()
# Check for exceptions
for task in done:
try:
task.result()
except Exception as e:
logger.error(f"Task error: {str(e)}")
except Exception as e:
logger.error(f"Connection error: {str(e)}")
async def forward_stdin_to_websocket(websocket):
"""Forward messages from stdin to the WebSocket server."""
loop = asyncio.get_event_loop()
while True:
# Read a line from stdin (non-blocking)
line = await loop.run_in_executor(None, sys.stdin.readline)
if not line:
logger.info("End of stdin, closing connection")
break
# Parse and forward the message
try:
message = line.strip()
logger.debug(f"Sending to WS: {message}")
await websocket.send(message)
except Exception as e:
logger.error(f"Error forwarding stdin to WebSocket: {str(e)}")
break
async def forward_websocket_to_stdout(websocket):
"""Forward messages from the WebSocket server to stdout."""
try:
async for message in websocket:
logger.debug(f"Received from WS: {message}")
# Write to stdout and flush
print(message, flush=True)
except ConnectionClosed:
logger.info("WebSocket connection closed")
except Exception as e:
logger.error(f"Error forwarding WebSocket to stdout: {str(e)}")
if __name__ == "__main__":
try:
# Run the message forwarding loop
asyncio.run(forward_messages())
except KeyboardInterrupt:
logger.info("Adapter terminated by user")
except Exception as e:
logger.error(f"Unhandled exception: {str(e)}")
sys.exit(1)
```
--------------------------------------------------------------------------------
/ai/PROMPTS.md:
--------------------------------------------------------------------------------
```markdown
# RhinoMCP Prompt Engineering Guide
This document provides guidance on effective prompt templates and strategies when using Claude AI with RhinoMCP for 3D modeling tasks.
## Overview
RhinoMCP exposes Rhino3D functionality to Claude through the Model Context Protocol, allowing the AI to create and manipulate 3D geometry. Effective prompts help Claude understand the design intent and execute the appropriate commands.
## Prompt Templates
### Basic Curve Creation
```
Create a NURBS curve in Rhino with the following points:
- Point 1: (0, 0, 0)
- Point 2: (5, 10, 0)
- Point 3: (10, 0, 0)
- Point 4: (15, 5, 0)
```
### Running Python Script in Rhino
```
Execute the following Python script in Rhino to [describe purpose]:
```python
# Your Python code here
import Rhino
import rhinoscriptsyntax as rs
# Example: Create a sphere
rs.AddSphere([0,0,0], 5)
```
```
### Checking Rhino Status
```
Check if Rhino is connected and report back with the version information.
```
## Prompt Strategies
### Be Specific with Coordinates
When creating geometry, provide exact coordinates rather than descriptive positions. This reduces ambiguity and ensures precise results.
**Good:**
```
Create a curve from (0,0,0) to (10,10,0) to (20,0,0)
```
**Less Effective:**
```
Create a curve that starts at the origin, goes up and to the right, then back down
```
### Use Visual References
For complex shapes, provide a visual reference or description to help Claude understand the desired outcome.
```
Create a curve that forms an "S" shape in the XY plane, starting at (0,0,0)
and ending at (20,0,0), with control points approximately at:
- (0,0,0)
- (5,10,0)
- (15,-10,0)
- (20,0,0)
```
### Iterate and Refine
Start with simple commands and gradually refine the design through conversation.
```
Step 1: Create a basic curve along the X-axis from (0,0,0) to (20,0,0)
Step 2: Now modify that curve to have a height of 5 units at its midpoint
Step 3: Create a second curve parallel to the first, offset by 10 units in the Y direction
```
### Specify Units
Always specify units when relevant to ensure the correct scale.
```
Create a sphere with radius 5 millimeters at position (10,10,10)
```
## Common Patterns
### Design Iteration
```
1. Create a basic version of [design element]
2. Evaluate the result
3. Modify specific aspects: "Make this part more [attribute]"
4. Continue refining until satisfied
```
### Reference-Based Design
```
1. Begin with a reference: "Create a curve similar to [description]"
2. Provide feedback on how to adjust
3. Add features or modify to match requirements
```
## Input Constraints
- Coordinate values should be numeric (avoid descriptive terms like "a little to the left")
- Python scripts must be compatible with Rhino's Python environment
- Avoid requesting operations on nonexistent objects
## Examples of Effective Prompts
### Example 1: Simple Curve
```
Create a NURBS curve in Rhino that forms a simple arc in the XY plane.
Use these points:
- (0,0,0)
- (5,5,0)
- (10,0,0)
```
### Example 2: Script-Based Operation
```
I want to create a grid of spheres in Rhino. Please execute a Python script that:
1. Creates a 3x3 grid of spheres
2. Sets the grid spacing to 10 units
3. Makes each sphere have a radius of 2 units
```
### Example 3: Multiple Operations
```
Let's create a simple model of a wine glass:
1. First create a vertical curve for the profile
2. Then create a circle at the base for the foot
3. Finally, use the existing Rhino commands to revolve the profile curve around the vertical axis
```
```
--------------------------------------------------------------------------------
/docs/ARCHITECTURE.md:
--------------------------------------------------------------------------------
```markdown
# RhinoMCP Architecture
This document outlines the architecture and component interactions of the RhinoMCP system, which connects Rhino3D to Claude AI via the Model Context Protocol.
## System Overview
RhinoMCP consists of three main components that work together to enable AI-assisted 3D modeling:
1. **Rhino Plugin**: A socket server running inside Rhino's Python environment
2. **Rhino Client**: A Python client that communicates with the Rhino plugin
3. **MCP Server**: A WebSocket server implementing the Model Context Protocol
These components interact in the following way:
```
Claude AI (Desktop/Windsurf) <--> MCP Server <--> Rhino Client <--> Rhino Plugin <--> Rhino3D
```
## Component Architecture
### 1. Rhino Plugin (`src/rhino_plugin/`)
The Rhino Plugin is a socket server that runs inside Rhino's Python editor environment. It serves as the interface to Rhino's functionality.
#### Key Files:
- `rhino_server.py`: Socket server implementation that listens for commands and executes them in Rhino
#### Responsibilities:
- Accept socket connections from external Python processes
- Receive commands in JSON format
- Execute commands in Rhino's context
- Return results or errors in JSON format
- Manage error handling and recovery
### 2. Rhino Client (`src/rhino_mcp/rhino_client.py`)
The Rhino Client is a Python module that communicates with the Rhino Plugin via a socket connection.
#### Responsibilities:
- Establish and maintain socket connection with the Rhino Plugin
- Format commands as JSON messages
- Send commands to the Rhino Plugin
- Receive and parse responses
- Provide a clean API for the MCP Server
### 3. MCP Server (`src/rhino_mcp/mcp_server.py`)
The MCP Server implements the Model Context Protocol and exposes Rhino functionality as MCP tools.
#### Responsibilities:
- Implement WebSocket server for MCP communication
- Register available Rhino tools
- Validate tool parameters
- Forward tool invocations to the Rhino Client
- Format responses according to MCP specifications
## Data Flow
1. **MCP Request Flow**:
- Claude AI sends a tool invocation request to the MCP Server
- MCP Server validates the request and parameters
- MCP Server formats the request for the Rhino Client
- Rhino Client sends the request to the Rhino Plugin
- Rhino Plugin executes the requested operation in Rhino
- Results flow back through the same path
2. **Communication Formats**:
- MCP Server <-> Claude AI: JSON-RPC over WebSockets
- Rhino Client <-> Rhino Plugin: Custom JSON protocol over TCP sockets
## Error Handling
The system implements multiple levels of error handling:
1. **MCP Server**: Validates requests and parameters before forwarding
2. **Rhino Client**: Handles connection issues and timeouts
3. **Rhino Plugin**: Catches exceptions during command execution in Rhino
4. **All Components**: Provide detailed error messages with stack traces for debugging
## Extensibility
The architecture is designed for extensibility:
1. **Tool Registration**: New tools can be added to the MCP Server without modifying the core code
2. **Command Handlers**: The Rhino Plugin can be extended with new command handlers
3. **Protocol Versioning**: Both socket protocols include version information for compatibility
## Security Considerations
1. **Local Connections Only**: Both socket servers bind to localhost by default
2. **No Authentication**: The current implementation assumes a trusted local environment
3. **Input Validation**: All component interfaces validate input to prevent injection attacks
## Future Considerations
1. **Geometry Transfer**: Optimize large geometry data transfer between components
2. **Connection Recovery**: Improve automatic reconnection for better resilience
3. **Tool Expansion**: Add more Rhino operations as MCP tools
4. **Authentication**: Add authentication for non-local deployments
```
--------------------------------------------------------------------------------
/tests/rhino_mcp/test_rhino_client.py:
--------------------------------------------------------------------------------
```python
"""Tests for the Rhino client module."""
from typing import Dict, Any, Optional, List, Tuple, Union, TypedDict, Generator
import socket
import threading
import time
import json
import pytest
from pytest.monkeypatch import MonkeyPatch
from pytest.logging import LogCaptureFixture
from rhino_mcp.rhino_client import RhinoClient, Point3d
class MockSocket:
"""Mock socket for testing."""
def __init__(self) -> None:
"""Initialize mock socket."""
self.sent_data: List[bytes] = []
self.responses: List[bytes] = []
def connect(self, addr: Tuple[str, int]) -> None:
"""Mock connect method."""
pass
def sendall(self, data: bytes) -> None:
"""Mock sendall method to record sent data."""
self.sent_data.append(data)
def recv(self, bufsize: int) -> bytes:
"""Mock recv method to return pre-configured responses."""
if not self.responses:
return b""
return self.responses.pop(0)
def close(self) -> None:
"""Mock close method."""
pass
def add_response(self, response: Dict[str, Any]) -> None:
"""Add a response to the mock socket."""
self.responses.append(json.dumps(response).encode('utf-8'))
@pytest.fixture
def mock_socket(monkeypatch: MonkeyPatch) -> Generator[MockSocket, None, None]:
"""Create a mock socket for testing."""
mock = MockSocket()
# Mock socket.socket to return our mock
def mock_socket_constructor(*args: Any, **kwargs: Any) -> MockSocket:
return mock
monkeypatch.setattr(socket, "socket", mock_socket_constructor)
yield mock
def test_rhino_client_connect(mock_socket: MockSocket) -> None:
"""Test RhinoClient connect method."""
client = RhinoClient()
# Test successful connection
assert client.connect() is True
assert client.connected is True
def test_rhino_client_ping(mock_socket: MockSocket) -> None:
"""Test RhinoClient ping method."""
client = RhinoClient()
client.connect()
# Add mock response for ping
mock_socket.add_response({
'status': 'success',
'message': 'Rhino is connected',
'data': {
'version': '8.0.0',
'has_active_doc': True,
'server_version': 'RhinoMCP-1.0',
'server_start_time': '2025-03-13 21:00:00',
'script_path': 'C:\\path\\to\\rhino_server.py'
}
})
# Test ping
response = client.ping()
# Verify request was sent
assert len(mock_socket.sent_data) == 1
request = json.loads(mock_socket.sent_data[0].decode('utf-8'))
assert request['type'] == 'ping'
# Verify response was parsed correctly
assert response['status'] == 'success'
assert response['message'] == 'Rhino is connected'
assert 'data' in response
assert response['data']['version'] == '8.0.0'
def test_rhino_client_create_curve(mock_socket: MockSocket) -> None:
"""Test RhinoClient create_curve method."""
client = RhinoClient()
client.connect()
# Add mock response for create_curve
mock_socket.add_response({
'status': 'success',
'message': 'Curve created with 3 points',
'data': {
'id': '12345-67890',
'point_count': 3
}
})
# Test points
points: List[Point3d] = [
{'x': 0.0, 'y': 0.0, 'z': 0.0},
{'x': 5.0, 'y': 10.0, 'z': 0.0},
{'x': 10.0, 'y': 0.0, 'z': 0.0}
]
# Test create_curve
response = client.create_curve(points)
# Verify request was sent
assert len(mock_socket.sent_data) == 1
request = json.loads(mock_socket.sent_data[0].decode('utf-8'))
assert request['type'] == 'create_curve'
assert 'data' in request
assert 'points' in request['data']
assert len(request['data']['points']) == 3
# Verify response was parsed correctly
assert response['status'] == 'success'
assert response['message'] == 'Curve created with 3 points'
assert 'data' in response
assert response['data']['id'] == '12345-67890'
def test_rhino_client_invalid_points() -> None:
"""Test RhinoClient create_curve with invalid points."""
client = RhinoClient()
# Test with empty points list
with pytest.raises(ValueError):
client.create_curve([])
# Test with single point
with pytest.raises(ValueError):
client.create_curve([{'x': 0.0, 'y': 0.0, 'z': 0.0}])
```
--------------------------------------------------------------------------------
/test_mcp_client.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""
Test MCP Client - Simple websocket client to test the RhinoMCP server.
This script simulates a client (like Claude) making requests to the RhinoMCP server
using the Model Context Protocol.
"""
import asyncio
import json
import websockets
from typing import Dict, Any, List, Optional
import logging
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-test-client")
# MCP server URI
MCP_URI = "ws://127.0.0.1:5000"
async def test_mcp_connection() -> bool:
"""Test connection to the MCP server.
Returns:
bool: True if the test passes, False otherwise
"""
try:
async with websockets.connect(MCP_URI) as websocket:
logger.info(f"Connected to MCP server at {MCP_URI}")
# Send initialize request
initialize_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {}
}
await websocket.send(json.dumps(initialize_request))
logger.info(f"Sent initialize request")
# Receive response
response = await websocket.recv()
initialize_response = json.loads(response)
logger.info(f"Received initialize response: {json.dumps(initialize_response, indent=2)}")
# Send initialized notification
initialized_notification = {
"jsonrpc": "2.0",
"method": "notifications/initialized"
}
await websocket.send(json.dumps(initialized_notification))
logger.info(f"Sent initialized notification")
# There may be a capabilities notification response, try to receive it
try:
response = await asyncio.wait_for(websocket.recv(), timeout=0.5)
capabilities_notification = json.loads(response)
logger.info(f"Received notification: {json.dumps(capabilities_notification, indent=2)}")
except asyncio.TimeoutError:
logger.info("No notification received (as expected)")
# Request tools list
tools_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
await websocket.send(json.dumps(tools_request))
logger.info(f"Sent tools/list request")
# Receive tools list response
response = await websocket.recv()
tools_response = json.loads(response)
logger.info(f"Received tools list: {json.dumps(tools_response, indent=2)}")
# Create a curve using the MCP protocol
create_curve_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "rhino_create_curve",
"params": {
"points": [
{"x": 0, "y": 0, "z": 30},
{"x": 10, "y": 10, "z": 30},
{"x": 20, "y": 0, "z": 30}
]
}
}
await websocket.send(json.dumps(create_curve_request))
logger.info(f"Sent curve creation request")
# Receive curve creation response
response = await websocket.recv()
curve_response = json.loads(response)
logger.info(f"Received curve creation response: {json.dumps(curve_response, indent=2)}")
# Check if curve was created successfully
if "result" in curve_response and "status" in curve_response["result"]:
status = curve_response["result"]["status"]
logger.info(f"Curve creation status: {status}")
if status == "success":
logger.info("✅ Test passed: Curve created successfully!")
return True
else:
logger.error(f"❌ Test failed: {curve_response['result']['message']}")
return False
else:
logger.error("❌ Test failed: Invalid response format")
return False
except Exception as e:
logger.error(f"❌ Error connecting to MCP server: {str(e)}")
return False
if __name__ == "__main__":
asyncio.run(test_mcp_connection())
```
--------------------------------------------------------------------------------
/claude_adapter.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""
RhinoMCP Claude Desktop Adapter - Bridges between Claude Desktop and Rhino
This adapter provides the stdio interface that Claude Desktop expects,
while communicating with a Rhino socket server.
"""
import sys
import json
import traceback
import asyncio
from rhino_mcp.rhino_client import RhinoClient
# MCP protocol version
PROTOCOL_VERSION = "2024-11-05"
# Global state
rhino_client = None
async def main():
# Initialize MCP session
await send_response(0, {
"protocolVersion": PROTOCOL_VERSION,
"serverInfo": {"name": "rhinomcp", "version": "0.1.0"},
"capabilities": {
"tools": {}
}
})
# Connect to Rhino
global rhino_client
rhino_client = RhinoClient()
try:
# Try to connect to Rhino
if not rhino_client.connect():
await send_log("error", "Failed to connect to Rhino server. Is it running?")
return
# Successfully connected
await send_log("info", f"Connected to Rhino server at {rhino_client.host}:{rhino_client.port}")
# Process incoming messages
while True:
# Read a line from stdin
line = await read_line()
if not line:
break
# Parse the message
try:
message = json.loads(line)
# Handle the message
await handle_message(message)
except json.JSONDecodeError:
await send_log("error", f"Invalid JSON: {line}")
except Exception as e:
await send_log("error", f"Error handling message: {str(e)}")
traceback.print_exc(file=sys.stderr)
finally:
# Disconnect from Rhino
if rhino_client and rhino_client.connected:
rhino_client.disconnect()
async def read_line():
"""Read a line from stdin asynchronously"""
return await asyncio.get_event_loop().run_in_executor(None, sys.stdin.readline)
async def send_message(message):
"""Send a message to stdout"""
print(json.dumps(message), flush=True)
async def send_response(id, result):
"""Send a JSON-RPC response"""
await send_message({
"jsonrpc": "2.0",
"id": id,
"result": result
})
async def send_error(id, code, message, data=None):
"""Send a JSON-RPC error response"""
error = {
"code": code,
"message": message
}
if data:
error["data"] = data
await send_message({
"jsonrpc": "2.0",
"id": id,
"error": error
})
async def send_log(level, message):
"""Send a log message notification"""
await send_message({
"jsonrpc": "2.0",
"method": "logging/message",
"params": {
"level": level,
"data": message
}
})
async def handle_message(message):
"""Handle an incoming JSON-RPC message"""
if "method" not in message:
await send_log("error", "Invalid message: no method")
return
method = message.get("method")
params = message.get("params", {})
msg_id = message.get("id")
if method == "initialize":
# Already sent in main()
pass
elif method == "initialized":
# Client is initialized, nothing to do
await send_log("info", "Client initialized")
elif method == "tools/list":
# Return list of available tools
await send_response(msg_id, {
"tools": [{
"name": "rhino_create_curve",
"description": "Create a NURBS curve in Rhino",
"inputSchema": {
"type": "object",
"properties": {
"points": {
"type": "array",
"description": "Array of 3D points for the curve",
"items": {
"type": "object",
"properties": {
"x": {"type": "number"},
"y": {"type": "number"},
"z": {"type": "number"}
},
"required": ["x", "y", "z"]
},
"minItems": 2
}
},
"required": ["points"]
}
}]
})
elif method == "tools/call":
tool_name = params.get("name")
tool_args = params.get("arguments", {})
if tool_name == "rhino_create_curve":
# Call the Rhino client to create a curve
try:
points = tool_args.get("points", [])
if not points or len(points) < 2:
await send_error(msg_id, -32602, "At least 2 points are required for a curve")
return
# Create the curve
result = rhino_client.create_curve(points)
if result.get("status") == "success":
await send_response(msg_id, {
"content": [{
"type": "text",
"text": f"Curve created successfully: {result.get('message', '')}"
}]
})
else:
await send_response(msg_id, {
"isError": True,
"content": [{
"type": "text",
"text": f"Failed to create curve: {result.get('message', 'Unknown error')}"
}]
})
except Exception as e:
await send_error(msg_id, -32603, f"Internal error: {str(e)}")
else:
await send_error(msg_id, -32601, f"Method not found: {tool_name}")
elif method == "exit":
# Client wants to exit
await send_log("info", "Shutting down")
sys.exit(0)
else:
await send_error(msg_id, -32601, f"Method not found: {method}")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
sys.exit(0)
except Exception as e:
print(f"Fatal error: {str(e)}", file=sys.stderr)
traceback.print_exc(file=sys.stderr)
sys.exit(1)
```
--------------------------------------------------------------------------------
/src/rhino_mcp/rhino_client.py:
--------------------------------------------------------------------------------
```python
"""
Rhino Client - Client for communicating with Rhino via socket connection.
This module implements a client that connects to the Rhino Bridge socket server
and provides methods to send commands and receive responses.
Version: 1.0 (2025-03-13)
"""
from typing import Dict, Any, Optional, List, Tuple, Union, TypedDict
import socket
import json
import time
import threading
from dataclasses import dataclass
class Point3d(TypedDict):
"""Type definition for a 3D point with x, y, z coordinates."""
x: float
y: float
z: float
class RhinoClient:
"""Client for communicating with Rhino via socket connection.
This class provides methods to connect to the Rhino Bridge socket server,
send commands, and receive responses.
Attributes:
host: The hostname or IP address of the Rhino Bridge server
port: The port number of the Rhino Bridge server
socket: The socket connection to the Rhino Bridge server
connected: Whether the client is currently connected to the server
"""
def __init__(self, host: str = '127.0.0.1', port: int = 8888):
"""Initialize the Rhino client.
Args:
host: The hostname or IP address of the Rhino Bridge server
port: The port number of the Rhino Bridge server
"""
self.host = host
self.port = port
self.socket: Optional[socket.socket] = None
self.connected = False
def connect(self) -> bool:
"""Connect to the Rhino Bridge server.
Returns:
True if connected successfully, False otherwise
Raises:
ConnectionError: If failed to connect to the server
"""
try:
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.connect((self.host, self.port))
self.connected = True
print(f"Connected to Rhino Bridge at {self.host}:{self.port}")
return True
except ConnectionRefusedError:
self.connected = False
raise ConnectionError(
f"Failed to connect to Rhino Bridge at {self.host}:{self.port}. "
"Make sure Rhino is running and the Bridge server is started."
)
except Exception as e:
self.connected = False
raise ConnectionError(f"Connection error: {str(e)}")
def disconnect(self) -> None:
"""Disconnect from the Rhino Bridge server.
Returns:
None
"""
if self.socket:
self.socket.close()
self.socket = None
self.connected = False
print("Disconnected from Rhino Bridge")
def send_command(self, cmd_type: str, data: Dict[str, Any] = None) -> Dict[str, Any]:
"""Send a command to the Rhino Bridge server.
Args:
cmd_type: The type of command to send
data: The data to send with the command
Returns:
The response from the server as a dictionary
Raises:
ConnectionError: If not connected to the server
RuntimeError: If failed to send command or receive response
"""
if not self.connected or not self.socket:
raise ConnectionError("Not connected to Rhino Bridge")
if data is None:
data = {}
try:
# Prepare the command
command = {
'type': cmd_type,
'data': data
}
# Send the command
self.socket.sendall(json.dumps(command).encode('utf-8'))
# Receive the response
response_data = self.socket.recv(4096)
if not response_data:
raise RuntimeError("No response from server")
# Parse the response
response = json.loads(response_data.decode('utf-8'))
return response
except Exception as e:
raise RuntimeError(f"Command error: {str(e)}")
def ping(self) -> Dict[str, Any]:
"""Ping the Rhino Bridge server to check connection.
Returns:
Server information including version and status
Raises:
ConnectionError: If not connected to the server
"""
return self.send_command('ping')
def create_curve(self, points: List[Point3d]) -> Dict[str, Any]:
"""Create a NURBS curve in Rhino.
Args:
points: List of points (each a dict with x, y, z keys)
Returns:
Response from the server including curve ID if successful
Raises:
ValueError: If points list is invalid
ConnectionError: If not connected to the server
"""
if not points or len(points) < 2:
raise ValueError("At least 2 points are required to create a curve")
return self.send_command('create_curve', {'points': points})
def refresh_view(self) -> Dict[str, Any]:
"""Refresh the Rhino viewport.
Returns:
Response from the server
Raises:
ConnectionError: If not connected to the server
"""
return self.send_command('refresh_view')
def run_script(self, script: str) -> Dict[str, Any]:
"""Run a Python script in Rhino's Python context.
Args:
script: The Python script to run
Returns:
Response from the server including script result
Raises:
ValueError: If script is empty
ConnectionError: If not connected to the server
"""
if not script:
raise ValueError("Script cannot be empty")
return self.send_command('run_script', {'script': script})
def test_connection(host: str = '127.0.0.1', port: int = 8888) -> bool:
"""Test the connection to the Rhino Bridge server.
Args:
host: The hostname or IP address of the Rhino Bridge server
port: The port number of the Rhino Bridge server
Returns:
True if connected successfully, False otherwise
"""
client = RhinoClient(host, port)
try:
client.connect()
response = client.ping()
print(f"Connection successful! Server info:")
for key, value in response.get('data', {}).items():
print(f" {key}: {value}")
return True
except Exception as e:
print(f"Connection test failed: {str(e)}")
return False
finally:
client.disconnect()
if __name__ == "__main__":
# Simple test when run directly
test_connection()
```
--------------------------------------------------------------------------------
/src/rhino_plugin/rhino_server.py:
--------------------------------------------------------------------------------
```python
"""
Rhino Bridge Server - Socket server for Rhino-Python communication.
This module implements a socket server that runs inside Rhino's Python editor and
allows external Python processes to communicate with and control Rhino.
Version: 1.0 (2025-03-13)
"""
from typing import Dict, Any, Optional, List, Tuple, Union
import socket
import json
import sys
import traceback
import threading
import time
# Import Rhino-specific modules
try:
import Rhino
import rhinoscriptsyntax as rs
import scriptcontext as sc
except ImportError:
print("Warning: Rhino modules not found. This script must run inside Rhino's Python editor.")
# Server configuration
HOST = '127.0.0.1'
PORT = 8888
SERVER_VERSION = "RhinoMCP-1.0"
SERVER_START_TIME = time.strftime("%Y-%m-%d %H:%M:%S")
class RhinoEncoder(json.JSONEncoder):
"""Custom JSON encoder for handling Rhino and .NET objects.
Args:
None
Returns:
Encoded JSON string
"""
def default(self, obj: Any) -> Any:
# Handle .NET Version objects and other common types
try:
if hasattr(obj, 'ToString'):
return str(obj)
elif hasattr(obj, 'Count') and hasattr(obj, 'Item'):
return [self.default(obj.Item[i]) for i in range(obj.Count)]
else:
return str(obj) # Last resort: convert anything to string
except:
return str(obj) # Absolute fallback
# Let the base class handle other types
return super(RhinoEncoder, self).default(obj)
def handle_client(conn: socket.socket, addr: Tuple[str, int]) -> None:
"""Handle individual client connections.
Args:
conn: Socket connection object
addr: Client address tuple (host, port)
Returns:
None
"""
print(f"Connection established from {addr}")
try:
while True:
# Receive command
data = conn.recv(4096)
if not data:
break
# Parse the command
try:
command_obj = json.loads(data.decode('utf-8'))
cmd_type = command_obj.get('type', '')
cmd_data = command_obj.get('data', {})
print(f"Received command: {cmd_type}")
result: Dict[str, Any] = {'status': 'error', 'message': 'Unknown command'}
# Process different command types
if cmd_type == 'ping':
result = {
'status': 'success',
'message': 'Rhino is connected',
'data': {
'version': str(Rhino.RhinoApp.Version),
'has_active_doc': Rhino.RhinoDoc.ActiveDoc is not None,
'server_version': SERVER_VERSION,
'server_start_time': SERVER_START_TIME,
'script_path': __file__
}
}
elif cmd_type == 'create_curve':
try:
# Extract points from the command data
points_data = cmd_data.get('points', [])
# Check if we have enough points for a curve
if len(points_data) < 2:
raise ValueError("At least 2 points are required to create a curve")
# Convert point data to Rhino points
points = []
for pt in points_data:
x = pt.get('x', 0.0)
y = pt.get('y', 0.0)
z = pt.get('z', 0.0)
points.append(Rhino.Geometry.Point3d(x, y, z))
# Create the curve
doc = Rhino.RhinoDoc.ActiveDoc
if not doc:
raise Exception("No active Rhino document")
# Create a NURBS curve
curve = Rhino.Geometry.Curve.CreateInterpolatedCurve(points, 3)
if not curve:
raise Exception("Failed to create curve")
# Add to document
id = doc.Objects.AddCurve(curve)
# Force view update
doc.Views.Redraw()
result = {
'status': 'success',
'message': f'Curve created with {len(points)} points',
'data': {
'id': str(id),
'point_count': len(points)
}
}
except Exception as e:
result = {
'status': 'error',
'message': f'Curve creation error: {str(e)}',
'traceback': traceback.format_exc()
}
elif cmd_type == 'refresh_view':
try:
doc = Rhino.RhinoDoc.ActiveDoc
if not doc:
raise Exception("No active Rhino document")
doc.Views.Redraw()
result = {
'status': 'success',
'message': 'View refreshed'
}
except Exception as e:
result = {
'status': 'error',
'message': f'View refresh error: {str(e)}'
}
elif cmd_type == 'run_script':
try:
script = cmd_data.get('script', '')
if not script:
raise ValueError("Empty script")
# Execute the script in Rhino's Python context
locals_dict = {}
exec(script, globals(), locals_dict)
# Return the result if available
script_result = locals_dict.get('result', None)
result = {
'status': 'success',
'message': 'Script executed successfully',
'data': {
'result': script_result
}
}
except Exception as e:
result = {
'status': 'error',
'message': f'Script execution error: {str(e)}',
'traceback': traceback.format_exc()
}
# Send the result back to the client
response = json.dumps(result, cls=RhinoEncoder)
conn.sendall(response.encode('utf-8'))
except json.JSONDecodeError:
conn.sendall(json.dumps({
'status': 'error',
'message': 'Invalid JSON format'
}).encode('utf-8'))
except Exception as e:
print(f"Connection error: {str(e)}")
finally:
print(f"Connection closed with {addr}")
conn.close()
def start_server() -> None:
"""Start the socket server.
Args:
None
Returns:
None
Raises:
OSError: If the server fails to start
"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
s.bind((HOST, PORT))
s.listen(5)
print(f"Rhino Bridge started! Listening on {HOST}:{PORT}")
print(f"Server version: {SERVER_VERSION}")
print(f"Start time: {SERVER_START_TIME}")
while True:
conn, addr = s.accept()
# Handle each client in a separate thread
client_thread = threading.Thread(target=handle_client, args=(conn, addr))
client_thread.daemon = True
client_thread.start()
except OSError as e:
if e.errno == 10048: # Address already in use
print("Error: Address already in use. Is the Rhino Bridge already running?")
else:
print(f"Socket error: {str(e)}")
except Exception as e:
print(f"Server error: {str(e)}")
traceback.print_exc()
if __name__ == "__main__":
# Only start if running directly in Rhino's Python editor
if 'Rhino' in sys.modules:
start_server()
else:
print("This script should be run inside Rhino's Python editor.")
```
--------------------------------------------------------------------------------
/src/rhino_mcp/mcp_server.py:
--------------------------------------------------------------------------------
```python
"""
MCP Server - Model Context Protocol server for RhinoMCP.
This module implements the Model Context Protocol server that connects
Claude AI to the Rhino client, enabling AI-assisted 3D modeling.
Version: 1.0 (2025-03-13)
"""
from typing import Dict, Any, Optional, List, Union, Callable, TypedDict
import os
import sys
import json
import logging
import argparse
from dataclasses import dataclass, field
import traceback
import asyncio
import websockets
from websockets.server import WebSocketServerProtocol
from rhino_mcp.rhino_client import RhinoClient, Point3d
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger("rhino_mcp")
class MCPRequestSchema(TypedDict):
"""Type definition for MCP request schema."""
jsonrpc: str
id: Union[str, int]
method: str
params: Dict[str, Any]
class MCPResponseSchema(TypedDict):
"""Type definition for MCP response schema."""
jsonrpc: str
id: Union[str, int]
result: Dict[str, Any]
class MCPErrorSchema(TypedDict):
"""Type definition for MCP error response schema."""
jsonrpc: str
id: Union[str, int]
error: Dict[str, Any]
@dataclass
class MCPTool:
"""Class representing an MCP tool.
Attributes:
name: The name of the tool
description: The description of the tool
parameters: The parameters schema of the tool
handler: The function to handle tool invocation
"""
name: str
description: str
parameters: Dict[str, Any]
handler: Callable[[Dict[str, Any]], Dict[str, Any]]
class RhinoMCPServer:
"""Model Context Protocol server for RhinoMCP.
This class implements the MCP server that handles communication with
Claude AI and forwards commands to the Rhino client.
Attributes:
host: The hostname to bind the server to
port: The port to bind the server to
rhino_client: The Rhino client to use for communication with Rhino
tools: The list of available MCP tools
"""
def __init__(
self,
host: str = '127.0.0.1',
port: int = 5000,
rhino_host: str = '127.0.0.1',
rhino_port: int = 8888
):
"""Initialize the MCP server.
Args:
host: The hostname to bind the server to
port: The port to bind the server to
rhino_host: The hostname of the Rhino Bridge server
rhino_port: The port of the Rhino Bridge server
"""
self.host = host
self.port = port
self.rhino_client = RhinoClient(rhino_host, rhino_port)
self.tools: List[MCPTool] = []
# Register built-in tools
self._register_tools()
def _register_tools(self) -> None:
"""Register built-in MCP tools.
This method registers the built-in MCP tools that will be exposed to
Claude AI through the Model Context Protocol.
Returns:
None
"""
# Create NURBS curve tool
self.tools.append(MCPTool(
name="rhino_create_curve",
description="Create a NURBS curve in Rhino",
parameters={
"type": "object",
"properties": {
"points": {
"type": "array",
"description": "Array of 3D points for the curve",
"items": {
"type": "object",
"properties": {
"x": {"type": "number"},
"y": {"type": "number"},
"z": {"type": "number"}
},
"required": ["x", "y", "z"]
},
"minItems": 2
}
},
"required": ["points"]
},
handler=self._handle_create_curve
))
# Tool for pinging Rhino
self.tools.append(MCPTool(
name="rhino_ping",
description="Ping Rhino to check if it's connected and get information",
parameters={
"type": "object",
"properties": {}
},
handler=self._handle_ping
))
# Tool for running Python script in Rhino
self.tools.append(MCPTool(
name="rhino_run_script",
description="Run a Python script in Rhino's Python context",
parameters={
"type": "object",
"properties": {
"script": {
"type": "string",
"description": "Python script to run in Rhino"
}
},
"required": ["script"]
},
handler=self._handle_run_script
))
def _handle_create_curve(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle create_curve tool invocation.
Args:
params: The parameters for the tool invocation
Returns:
The tool invocation result
Raises:
ValueError: If the parameters are invalid
"""
points_data = params.get('points', [])
# Validate points
if not points_data or len(points_data) < 2:
raise ValueError("At least 2 points are required to create a curve")
# Format points for Rhino client
points: List[Point3d] = []
for pt in points_data:
points.append({
'x': float(pt.get('x', 0.0)),
'y': float(pt.get('y', 0.0)),
'z': float(pt.get('z', 0.0))
})
# Ensure Rhino client is connected
if not self.rhino_client.connected:
self.rhino_client.connect()
# Create the curve
result = self.rhino_client.create_curve(points)
# Format response
if result.get('status') == 'success':
return {
'success': True,
'message': result.get('message', 'Curve created successfully'),
'data': result.get('data', {})
}
else:
return {
'success': False,
'message': result.get('message', 'Failed to create curve'),
'error': result.get('traceback', '')
}
def _handle_ping(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle ping tool invocation.
Args:
params: The parameters for the tool invocation
Returns:
The tool invocation result
"""
# Ensure Rhino client is connected
if not self.rhino_client.connected:
self.rhino_client.connect()
# Ping Rhino
result = self.rhino_client.ping()
# Format response
if result.get('status') == 'success':
return {
'success': True,
'message': result.get('message', 'Rhino is connected'),
'data': result.get('data', {})
}
else:
return {
'success': False,
'message': result.get('message', 'Failed to ping Rhino'),
'error': result.get('traceback', '')
}
def _handle_run_script(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle run_script tool invocation.
Args:
params: The parameters for the tool invocation
Returns:
The tool invocation result
Raises:
ValueError: If the script is empty
"""
script = params.get('script', '')
# Validate script
if not script:
raise ValueError("Script cannot be empty")
# Ensure Rhino client is connected
if not self.rhino_client.connected:
self.rhino_client.connect()
# Run the script
result = self.rhino_client.run_script(script)
# Format response
if result.get('status') == 'success':
return {
'success': True,
'message': result.get('message', 'Script executed successfully'),
'data': result.get('data', {})
}
else:
return {
'success': False,
'message': result.get('message', 'Failed to execute script'),
'error': result.get('traceback', '')
}
def get_tools_schema(self) -> List[Dict[str, Any]]:
"""Get the schema for all registered tools.
Returns:
List of tool schemas in MCP format
"""
return [
{
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters
}
for tool in self.tools
]
async def handle_jsonrpc(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""Handle a JSON-RPC request.
Args:
request: The JSON-RPC request
Returns:
The JSON-RPC response
"""
# Extract request data
method = request.get('method', '')
params = request.get('params', {})
req_id = request.get('id', 0)
# Handle different methods
if method == 'rpc.discover':
# Return MCP server information and tools
return {
'jsonrpc': '2.0',
'id': req_id,
'result': {
'name': 'rhino_mcp',
'version': '1.0.0',
'functions': self.get_tools_schema()
}
}
elif method.startswith('rhino_'):
# Handle tool invocation
for tool in self.tools:
if tool.name == method:
try:
result = tool.handler(params)
return {
'jsonrpc': '2.0',
'id': req_id,
'result': result
}
except Exception as e:
logger.error(f"Tool error: {str(e)}")
traceback.print_exc()
return {
'jsonrpc': '2.0',
'id': req_id,
'error': {
'code': -32000,
'message': str(e),
'data': {
'traceback': traceback.format_exc()
}
}
}
# Method not found
return {
'jsonrpc': '2.0',
'id': req_id,
'error': {
'code': -32601,
'message': f'Method not found: {method}'
}
}
async def handle_websocket(self, websocket: WebSocketServerProtocol) -> None:
"""Handle a WebSocket connection.
Args:
websocket: The WebSocket connection
Returns:
None
"""
logger.info(f"Client connected: {websocket.remote_address}")
# Ensure Rhino client is connected
if not self.rhino_client.connected:
try:
self.rhino_client.connect()
except Exception as e:
logger.error(f"Failed to connect to Rhino: {str(e)}")
await websocket.close(1011, "Failed to connect to Rhino")
return
try:
async for message in websocket:
# Parse the message
try:
request = json.loads(message)
logger.info(f"Received request: {request.get('method', 'unknown')}")
# Handle the request
response = await self.handle_jsonrpc(request)
# Send the response
await websocket.send(json.dumps(response))
except json.JSONDecodeError:
logger.error("Invalid JSON")
await websocket.send(json.dumps({
'jsonrpc': '2.0',
'id': None,
'error': {
'code': -32700,
'message': 'Parse error'
}
}))
except Exception as e:
logger.error(f"Websocket error: {str(e)}")
traceback.print_exc()
await websocket.send(json.dumps({
'jsonrpc': '2.0',
'id': None,
'error': {
'code': -32603,
'message': str(e)
}
}))
except Exception as e:
logger.error(f"Connection error: {str(e)}")
finally:
logger.info(f"Client disconnected: {websocket.remote_address}")
async def start(self) -> None:
"""Start the MCP server.
Returns:
None
"""
# Start the WebSocket server
async with websockets.serve(self.handle_websocket, self.host, self.port):
logger.info(f"MCP server started at ws://{self.host}:{self.port}")
await asyncio.Future() # Run forever
def start_in_thread(self) -> None:
"""Start the MCP server in a separate thread.
Returns:
None
"""
try:
asyncio.run(self.start())
except KeyboardInterrupt:
logger.info("Server stopped by user")
except Exception as e:
logger.error(f"Server error: {str(e)}")
traceback.print_exc()
finally:
if self.rhino_client.connected:
self.rhino_client.disconnect()
def main() -> None:
"""Start the MCP server from the command line.
Returns:
None
"""
parser = argparse.ArgumentParser(description='Start the RhinoMCP server')
parser.add_argument('--host', type=str, default='127.0.0.1',
help='Hostname to bind the MCP server to')
parser.add_argument('--port', type=int, default=5000,
help='Port to bind the MCP server to')
parser.add_argument('--rhino-host', type=str, default='127.0.0.1',
help='Hostname of the Rhino Bridge server')
parser.add_argument('--rhino-port', type=int, default=8888,
help='Port of the Rhino Bridge server')
parser.add_argument('--debug', action='store_true',
help='Enable debug logging')
args = parser.parse_args()
# Set log level
if args.debug:
logger.setLevel(logging.DEBUG)
# Start the server
server = RhinoMCPServer(
host=args.host,
port=args.port,
rhino_host=args.rhino_host,
rhino_port=args.rhino_port
)
print(f"Starting RhinoMCP server at ws://{args.host}:{args.port}")
print(f"Connecting to Rhino at {args.rhino_host}:{args.rhino_port}")
print("Press Ctrl+C to stop")
try:
server.start_in_thread()
except KeyboardInterrupt:
print("Server stopped by user")
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/ai/rhino-python-bridge-docs.md:
--------------------------------------------------------------------------------
```markdown
# Python-Rhino Bridge: Setup & Usage Guide
This guide explains how to establish a bidirectional connection between external Python scripts and Rhinoceros 3D, allowing you to control Rhino programmatically from Python running outside the Rhino environment.
## Overview
The Python-Rhino Bridge consists of two main components:
1. **Server Script** - Runs inside Rhino's Python editor and listens for commands
2. **Client Script** - Runs in your external Python environment and sends commands to Rhino
This connection enables:
- Creating Rhino geometry from external Python
- Running custom Python code within Rhino's environment
- Querying information from Rhino
- Building interactive tools that communicate with Rhino
## Requirements
### Software Requirements
- **Rhinoceros 3D**: Version 7 or 8
- **Python**: Version 3.9 or 3.10 (matching your Rhino installation)
- **Operating System**: Windows 10 or 11
### Directory Structure
Create a project folder with the following structure:
```
rhino_windsurf/
├── rhino_bridge.py # Server script (runs in Rhino)
├── rhino_client_deluxe.py # Client script (runs in external Python)
└── PythonBridgeCommand.py # Optional Rhino command script
```
## Setup Instructions
### 1. Setting Up the Rhino Server
1. **Create the server script**:
- Open Rhino
- Type `EditPythonScript` in the command line to open Rhino's Python editor
- Create a new script named `rhino_bridge.py`
- Copy the server script code (provided below) into this file
- Save the file to your project directory
2. **Running the server**:
- With the `rhino_bridge.py` file open in Rhino's Python editor
- Click the "Run" button or press F5
- Verify you see "Rhino Bridge started!" in the output panel
- Keep this script running as long as you need the connection
### 2. Setting Up the Python Client
1. **Python environment setup**:
- Create a virtual environment (recommended):
```
python -m venv .venv_rhino
.venv_rhino\Scripts\activate # On Windows
```
- This isolates your project dependencies from other Python projects
2. **Create the client script**:
- Create a new file named `rhino_client_deluxe.py` in your project directory
- Copy the client script code (provided below) into this file
3. **Running the client**:
- Ensure Rhino is running with the server script active
- Open a terminal in your project directory
- Activate your virtual environment if used
- Run: `python rhino_client_deluxe.py`
- You should see a confirmation of successful connection
### 3. Optional: Creating a Rhino Command
To start the server easily within Rhino:
1. Create `PythonBridgeCommand.py` in your Rhino scripts folder:
- Typically located at `%APPDATA%\McNeel\Rhinoceros\8.0\scripts\`
- Copy the command script code (provided below) into this file
2. Run the command in Rhino:
- Type `PythonBridge` in Rhino's command line
- The server should start automatically
## Script Code
### 1. Rhino Server Script (`rhino_bridge.py`)
```python
"""
Rhino Bridge - Simplified server for stable Python-Rhino communication.
Version: 2.0 (2025-03-13)
This script provides a reliable socket connection between Python and Rhino
with simplified error handling and robust object creation.
"""
import socket
import json
import sys
import traceback
import threading
import Rhino
import time
# Server configuration
HOST = '127.0.0.1'
PORT = 8888
SERVER_VERSION = "Bridge-2.0"
SERVER_START_TIME = time.strftime("%Y-%m-%d %H:%M:%S")
# Custom JSON encoder for handling .NET objects
class RhinoEncoder(json.JSONEncoder):
def default(self, obj):
# Handle .NET Version objects and other common types
try:
if hasattr(obj, 'ToString'):
return str(obj)
elif hasattr(obj, 'Count') and hasattr(obj, 'Item'):
return [self.default(obj.Item[i]) for i in range(obj.Count)]
else:
return str(obj) # Last resort: convert anything to string
except:
return str(obj) # Absolute fallback
# Let the base class handle other types
return super(RhinoEncoder, self).default(obj)
def handle_client(conn, addr):
"""Handle individual client connections"""
print(f"Connection established from {addr}")
try:
while True:
# Receive command
data = conn.recv(4096)
if not data:
break
# Parse the command
try:
command_obj = json.loads(data.decode('utf-8'))
cmd_type = command_obj.get('type', '')
cmd_data = command_obj.get('data', {})
print(f"Received command: {cmd_type}")
result = {'status': 'error', 'message': 'Unknown command'}
# Process different command types
if cmd_type == 'ping':
result = {
'status': 'success',
'message': 'Rhino is connected',
'data': {
'version': str(Rhino.RhinoApp.Version),
'has_active_doc': Rhino.RhinoDoc.ActiveDoc is not None,
'server_version': SERVER_VERSION,
'server_start_time': SERVER_START_TIME,
'script_path': __file__
}
}
elif cmd_type == 'create_sphere':
# SIMPLIFIED APPROACH: Just create the sphere without complex checks
try:
center_x = cmd_data.get('center_x', 0)
center_y = cmd_data.get('center_y', 0)
center_z = cmd_data.get('center_z', 0)
radius = cmd_data.get('radius', 5.0)
doc = Rhino.RhinoDoc.ActiveDoc
if not doc:
raise Exception("No active Rhino document")
# Create the sphere directly
center = Rhino.Geometry.Point3d(center_x, center_y, center_z)
sphere = Rhino.Geometry.Sphere(center, radius)
# Convert to a brep for better display
brep = sphere.ToBrep()
# Add to document, ignoring the return value
doc.Objects.AddBrep(brep)
# Force view update
doc.Views.Redraw()
result = {
'status': 'success',
'message': f'Sphere created at ({center_x}, {center_y}, {center_z}) with radius {radius}'
}
except Exception as e:
result = {
'status': 'error',
'message': f'Sphere creation error: {str(e)}',
'traceback': traceback.format_exc()
}
elif cmd_type == 'refresh_view':
try:
doc = Rhino.RhinoDoc.ActiveDoc
if doc:
doc.Views.Redraw()
result = {'status': 'success', 'message': 'Views refreshed'}
else:
result = {'status': 'error', 'message': 'No active document'}
except Exception as e:
result = {'status': 'error', 'message': f'Refresh error: {str(e)}'}
elif cmd_type == 'run_script':
script = cmd_data.get('script', '')
if script:
# Capture print output
old_stdout = sys.stdout
from io import StringIO
captured_output = StringIO()
sys.stdout = captured_output
try:
# Execute the script
exec(script)
result = {
'status': 'success',
'message': 'Script executed',
'data': {'output': captured_output.getvalue()}
}
except Exception as e:
result = {
'status': 'error',
'message': f'Script execution error: {str(e)}',
'data': {'traceback': traceback.format_exc()}
}
finally:
sys.stdout = old_stdout
else:
result = {'status': 'error', 'message': 'No script provided'}
# Send the result back using the custom encoder
conn.sendall(json.dumps(result, cls=RhinoEncoder).encode('utf-8'))
except json.JSONDecodeError:
conn.sendall(json.dumps({
'status': 'error',
'message': 'Invalid JSON format'
}).encode('utf-8'))
except Exception as e:
print(f"Error processing command: {str(e)}")
traceback.print_exc()
conn.sendall(json.dumps({
'status': 'error',
'message': f'Server error: {str(e)}',
'traceback': traceback.format_exc()
}, cls=RhinoEncoder).encode('utf-8'))
except Exception as e:
print(f"Connection error: {str(e)}")
finally:
print(f"Connection closed with {addr}")
conn.close()
def start_server():
"""Start the socket server"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
s.bind((HOST, PORT))
s.listen()
print(f"Server started on {HOST}:{PORT}")
print("Waiting for connections...")
try:
while True:
conn, addr = s.accept()
client_thread = threading.Thread(target=handle_client, args=(conn, addr))
client_thread.daemon = True
client_thread.start()
except KeyboardInterrupt:
print("Server shutting down...")
except Exception as e:
print(f"Server error: {str(e)}")
traceback.print_exc()
except Exception as e:
print(f"Failed to bind to {HOST}:{PORT}. Error: {str(e)}")
print("Try closing any other running instances of this script or check if another program is using this port.")
# Display server information
print("\n========== RHINO BRIDGE ==========")
print(f"Version: {SERVER_VERSION}")
print(f"Started: {SERVER_START_TIME}")
print(f"File: {__file__}")
print("===================================\n")
# Start the server in a background thread to keep Rhino responsive
server_thread = threading.Thread(target=start_server)
server_thread.daemon = True
server_thread.start()
print("Rhino Bridge started!")
print(f"Listening on {HOST}:{PORT}")
print("Keep this script running in Rhino's Python editor")
print("The server will run until you close this script or Rhino")
```
### 2. Python Client Script (`rhino_client_deluxe.py`)
```python
"""
Rhino Client Deluxe - Interactive client for Rhino connection.
Version: 1.0 (2025-03-13)
This script provides an interactive terminal for connecting to and
controlling Rhino from external Python scripts.
"""
import os
import sys
import json
import socket
import time
from typing import Dict, Any, Optional, List, Tuple
class RhinoClient:
"""Client for maintaining an interactive connection with Rhino."""
def __init__(self, host: str = '127.0.0.1', port: int = 8888):
"""Initialize the interactive Rhino client."""
self.host = host
self.port = port
self.socket = None
self.connected = False
self.command_history: List[str] = []
def connect(self) -> bool:
"""Establish connection to the Rhino socket server."""
try:
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.connect((self.host, self.port))
self.connected = True
# Test connection with ping and verify server version
response = self.send_command('ping')
if response and response.get('status') == 'success':
server_data = response.get('data', {})
print(f"\n✅ Connected to Rhino {server_data.get('version', 'unknown')}")
print(f"🔌 Server: {server_data.get('server_version', 'unknown')}")
print(f"📂 Script: {server_data.get('script_path', 'unknown')}")
print(f"⏰ Started: {server_data.get('server_start_time', 'unknown')}")
# Ensure we're connected to the Deluxe server
if 'Deluxe' not in server_data.get('server_version', ''):
print("\n⚠️ WARNING: Not connected to RhinoServerDeluxe!")
print("You may experience errors with spheres and other commands.")
print("Please run rhino_server_deluxe.py in Rhino's Python editor.")
# Add an immediate view refresh to ensure everything is visible
self.refresh_view()
return True
print("\n❌ Connection test failed")
self.disconnect()
return False
except Exception as e:
print(f"\n❌ Connection error: {str(e)}")
return False
def disconnect(self) -> None:
"""Close the connection to Rhino."""
if self.socket:
try:
self.socket.close()
except:
pass
self.socket = None
self.connected = False
print("\nDisconnected from Rhino")
def send_command(self, cmd_type: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Send a command to the Rhino server and return the response."""
if not self.connected or not self.socket:
print("❌ Not connected to Rhino")
return {'status': 'error', 'message': 'Not connected to Rhino'}
try:
# Create command
command = {
'type': cmd_type,
'data': data or {}
}
# Send command
self.socket.sendall(json.dumps(command).encode('utf-8'))
# Receive response
response = self.socket.recv(8192)
return json.loads(response.decode('utf-8'))
except Exception as e:
print(f"❌ Error sending command: {str(e)}")
# Don't disconnect automatically to allow for retry
return {'status': 'error', 'message': f'Command error: {str(e)}'}
def create_sphere(self, x: float, y: float, z: float, radius: float) -> Dict[str, Any]:
"""Create a sphere in Rhino."""
data = {
'center_x': x,
'center_y': y,
'center_z': z,
'radius': radius
}
result = self.send_command('create_sphere', data)
# Always refresh view after creating geometry
if result.get('status') == 'success':
self.refresh_view()
return result
def run_script(self, script: str) -> Dict[str, Any]:
"""Run a Python script in Rhino."""
result = self.send_command('run_script', {'script': script})
# Always refresh view after running a script
if result.get('status') == 'success':
self.refresh_view()
return result
def refresh_view(self) -> Dict[str, Any]:
"""Refresh the Rhino viewport."""
return self.send_command('refresh_view')
def add_to_history(self, command: str) -> None:
"""Add a command to the history."""
if command and command not in ['', 'exit', 'help']:
self.command_history.append(command)
if len(self.command_history) > 100: # Limit history size
self.command_history.pop(0)
def print_help() -> None:
"""Print help information about available commands."""
print("\n=== Available Commands ===")
print("sphere <x> <y> <z> <radius> - Create a sphere")
print("script <python_code> - Run Python code in Rhino")
print("refresh - Refresh the Rhino viewport")
print("history - Show command history")
print("ping - Test connection to Rhino")
print("help - Show this help message")
print("exit - Close the connection and exit")
print("\nExample: sphere 10 20 0 5")
print("Example: script import Rhino; print(f\"Current doc: {Rhino.RhinoDoc.ActiveDoc.Name}\")")
def main() -> None:
"""Run the interactive Rhino client."""
print("\n========== RHINO CLIENT DELUXE ==========")
print("Version: 1.0 (2025-03-13)")
print("==========================================\n")
print("This script provides an interactive connection to Rhino.")
print("Make sure 'rhino_server_deluxe.py' is running in Rhino's Python editor.")
# Create client
client = RhinoClient()
# Connect to Rhino
if not client.connect():
print("\nFailed to connect to Rhino. Make sure the server script is running.")
return
print_help()
# Command loop
try:
while True:
command = input("\nrhino> ").strip()
if not command:
continue
if command.lower() == 'exit':
break
if command.lower() == 'help':
print_help()
continue
if command.lower() == 'ping':
response = client.send_command('ping')
if response.get('status') == 'success':
print(f"Connection active! Rhino version: {response.get('data', {}).get('version', 'unknown')}")
else:
print(f"Ping failed: {response.get('message', 'Unknown error')}")
client.add_to_history(command)
continue
# View refresh command
if command.lower() == 'refresh':
response = client.refresh_view()
if response.get('status') == 'success':
print("✅ Viewport refreshed")
else:
print(f"❌ Refresh failed: {response.get('message', 'Unknown error')}")
client.add_to_history(command)
continue
# Command history
if command.lower() == 'history':
if client.command_history:
print("\n=== Command History ===")
for i, cmd in enumerate(client.command_history):
print(f"{i+1}. {cmd}")
else:
print("No command history yet.")
continue
# Parse sphere command: sphere <x> <y> <z> <radius>
if command.lower().startswith('sphere '):
try:
parts = command.split()
if len(parts) != 5:
print("❌ Invalid sphere command. Format: sphere <x> <y> <z> <radius>")
continue
x, y, z, radius = map(float, parts[1:])
response = client.create_sphere(x, y, z, radius)
if response.get('status') == 'success':
print(f"✅ {response.get('message', 'Sphere created successfully')}")
else:
print(f"❌ Sphere creation failed: {response.get('message', 'Unknown error')}")
if response.get('traceback'):
print("\nError details:")
print(response.get('traceback'))
print("\nTroubleshooting tip: Make sure you're running rhino_server_deluxe.py in Rhino")
client.add_to_history(command)
except ValueError:
print("❌ Invalid parameters. All values must be numbers.")
continue
# Handle script command: script <python code>
if command.lower().startswith('script '):
script = command[7:] # Remove 'script ' prefix
response = client.run_script(script)
if response.get('status') == 'success':
print("\n=== Script Output ===")
output = response.get('data', {}).get('output', 'No output')
if output.strip():
print(output)
else:
print("Script executed successfully (no output)")
else:
print(f"❌ Script error: {response.get('message', 'Unknown error')}")
if 'traceback' in response:
print("\n=== Error Traceback ===")
print(response['traceback'])
client.add_to_history(command)
continue
print(f"❌ Unknown command: {command}")
print("Type 'help' for available commands")
except KeyboardInterrupt:
print("\nInterrupted by user")
except Exception as e:
print(f"\n❌ Error: {str(e)}")
import traceback
traceback.print_exc()
finally:
client.disconnect()
if __name__ == "__main__":
main()
```
### 3. Rhino Command Script (`PythonBridgeCommand.py`)
```python
"""Rhino-Python Bridge Command
Creates a custom Rhino command to start the Python socket server.
"""
import Rhino
import rhinoscriptsyntax as rs
import scriptcontext as sc
import System
import os
__commandname__ = "PythonBridge"
def RunCommand(is_interactive):
"""Run the PythonBridge command, which starts the socket server for Python connections."""
# Create path to the user's Documents folder (reliable location)
docs_folder = System.Environment.GetFolderPath(System.Environment.SpecialFolder.MyDocuments)
projects_folder = os.path.join(docs_folder, "CascadeProjects", "rhino_windsurf")
# Path to the server script
script_path = os.path.join(projects_folder, "rhino_bridge.py")
# Check if the file exists at the default location
if not os.path.exists(script_path):
# If not found, ask the user for the file location
filter = "Python Files (*.py)|*.py|All Files (*.*)|*.*||"
script_path = rs.OpenFileName("Select the rhino_bridge.py script", filter)
if not script_path:
print("Operation canceled. No server script selected.")
return Rhino.Commands.Result.Cancel
# Run the script using Rhino's script runner
Rhino.RhinoApp.RunScript("_-RunPythonScript \"" + script_path + "\"", False)
print("Python Bridge server started!")
print(f"Server script: {script_path}")
print("You can now connect from external Python scripts")
return Rhino.Commands.Result.Success
# This is needed for Rhino to recognize the command
if __name__ == "__main__":
RunCommand(True)
```
## Using the Python-Rhino Bridge
### Basic Commands
Once the bridge is set up and both the server and client are running, you can interact with Rhino using these commands:
1. **Creating Spheres**:
```
rhino> sphere <x> <y> <z> <radius>
```
Example: `sphere 10 20 0 5`
2. **Running Python Scripts in Rhino**:
```
rhino> script <python_code>
```
Example: `script import Rhino; print(f"Rhino version: {Rhino.RhinoApp.Version}")`
3. **Refreshing the Viewport**:
```
rhino> refresh
```
4. **Testing Connection**:
```
rhino> ping
```
5. **Viewing Command History**:
```
rhino> history
```
6. **Exiting the Client**:
```
rhino> exit
```
### Advanced Usage: Multi-Statement Python Commands
You can run multiple Python statements in a single command by separating them with semicolons:
```
rhino> script import Rhino; import random; x = random.uniform(0,10); y = random.uniform(0,10); print(f"Random point: ({x}, {y})")
```
For more complex scripts, you can use Python's block syntax with proper indentation:
```
rhino> script if True:
import Rhino
import random
for i in range(3):
x = random.uniform(0, 10)
y = random.uniform(0, 10)
z = random.uniform(0, 10)
print(f"Point {i+1}: ({x}, {y}, {z})")
```
### Creating Complex Geometry
To create more complex geometry, you can write scripts that leverage the full power of RhinoCommon:
```
rhino> script import Rhino; doc = Rhino.RhinoDoc.ActiveDoc; curve = Rhino.Geometry.Circle(Rhino.Geometry.Plane.WorldXY, 10).ToNurbsCurve(); doc.Objects.AddCurve(curve); doc.Views.Redraw()
```
## Troubleshooting
### Common Issues and Solutions
1. **Connection Refused**:
- Ensure Rhino is running with the server script active
- Check if another instance of the server is already running on port 8888
- Try restarting Rhino and the server script
2. **Geometry Not Appearing**:
- Use the `refresh` command to force a viewport update
- Make sure you have an active document open in Rhino
- Check for error messages in the command response
3. **Script Execution Errors**:
- Verify your Python syntax is correct
- Make sure you're using RhinoCommon API methods correctly
- Check the error traceback for specific issues
4. **Server Warning in Client**:
- The "Not connected to RhinoServerDeluxe" warning can be ignored if using the Bridge version
- Make sure you're using compatible versions of the server and client scripts
### Extending the Bridge
You can extend the functionality of the bridge by:
1. Adding new command types to the server script
2. Implementing additional geometry creation methods in the client
3. Creating specialized scriptlets for common tasks
## Conclusion
The Python-Rhino Bridge provides a powerful way to control Rhino programmatically from external Python scripts. Whether you're automating modeling tasks, integrating with other tools, or building custom workflows, this bridge offers a flexible and robust connection between Python and Rhino.
```
--------------------------------------------------------------------------------
/rhino-python-bridge-docs.md:
--------------------------------------------------------------------------------
```markdown
# Python-Rhino Bridge: Setup & Usage Guide
This guide explains how to establish a bidirectional connection between external Python scripts and Rhinoceros 3D, allowing you to control Rhino programmatically from Python running outside the Rhino environment.
## Overview
The Python-Rhino Bridge consists of two main components:
1. **Server Script** - Runs inside Rhino's Python editor and listens for commands
2. **Client Script** - Runs in your external Python environment and sends commands to Rhino
This connection enables:
- Creating Rhino geometry from external Python
- Running custom Python code within Rhino's environment
- Querying information from Rhino
- Building interactive tools that communicate with Rhino
## Requirements
### Software Requirements
- **Rhinoceros 3D**: Version 7 or 8
- **Python**: Version 3.9 or 3.10 (matching your Rhino installation)
- **Operating System**: Windows 10 or 11
### Directory Structure
Create a project folder with the following structure:
```
rhino_windsurf/
├── rhino_bridge.py # Server script (runs in Rhino)
├── rhino_client_deluxe.py # Client script (runs in external Python)
└── PythonBridgeCommand.py # Optional Rhino command script
```
## Setup Instructions
### 1. Setting Up the Rhino Server
1. **Create the server script**:
- Open Rhino
- Type `EditPythonScript` in the command line to open Rhino's Python editor
- Create a new script named `rhino_bridge.py`
- Copy the server script code (provided below) into this file
- Save the file to your project directory
2. **Running the server**:
- With the `rhino_bridge.py` file open in Rhino's Python editor
- Click the "Run" button or press F5
- Verify you see "Rhino Bridge started!" in the output panel
- Keep this script running as long as you need the connection
### 2. Setting Up the Python Client
1. **Python environment setup**:
- Create a virtual environment (recommended):
```
python -m venv .venv_rhino
.venv_rhino\Scripts\activate # On Windows
```
- This isolates your project dependencies from other Python projects
2. **Create the client script**:
- Create a new file named `rhino_client_deluxe.py` in your project directory
- Copy the client script code (provided below) into this file
3. **Running the client**:
- Ensure Rhino is running with the server script active
- Open a terminal in your project directory
- Activate your virtual environment if used
- Run: `python rhino_client_deluxe.py`
- You should see a confirmation of successful connection
### 3. Optional: Creating a Rhino Command
To start the server easily within Rhino:
1. Create `PythonBridgeCommand.py` in your Rhino scripts folder:
- Typically located at `%APPDATA%\McNeel\Rhinoceros\8.0\scripts\`
- Copy the command script code (provided below) into this file
2. Run the command in Rhino:
- Type `PythonBridge` in Rhino's command line
- The server should start automatically
## Script Code
### 1. Rhino Server Script (`rhino_bridge.py`)
```python
"""
Rhino Bridge - Simplified server for stable Python-Rhino communication.
Version: 2.0 (2025-03-13)
This script provides a reliable socket connection between Python and Rhino
with simplified error handling and robust object creation.
"""
import socket
import json
import sys
import traceback
import threading
import Rhino
import time
# Server configuration
HOST = '127.0.0.1'
PORT = 8888
SERVER_VERSION = "Bridge-2.0"
SERVER_START_TIME = time.strftime("%Y-%m-%d %H:%M:%S")
# Custom JSON encoder for handling .NET objects
class RhinoEncoder(json.JSONEncoder):
def default(self, obj):
# Handle .NET Version objects and other common types
try:
if hasattr(obj, 'ToString'):
return str(obj)
elif hasattr(obj, 'Count') and hasattr(obj, 'Item'):
return [self.default(obj.Item[i]) for i in range(obj.Count)]
else:
return str(obj) # Last resort: convert anything to string
except:
return str(obj) # Absolute fallback
# Let the base class handle other types
return super(RhinoEncoder, self).default(obj)
def handle_client(conn, addr):
"""Handle individual client connections"""
print(f"Connection established from {addr}")
try:
while True:
# Receive command
data = conn.recv(4096)
if not data:
break
# Parse the command
try:
command_obj = json.loads(data.decode('utf-8'))
cmd_type = command_obj.get('type', '')
cmd_data = command_obj.get('data', {})
print(f"Received command: {cmd_type}")
result = {'status': 'error', 'message': 'Unknown command'}
# Process different command types
if cmd_type == 'ping':
result = {
'status': 'success',
'message': 'Rhino is connected',
'data': {
'version': str(Rhino.RhinoApp.Version),
'has_active_doc': Rhino.RhinoDoc.ActiveDoc is not None,
'server_version': SERVER_VERSION,
'server_start_time': SERVER_START_TIME,
'script_path': __file__
}
}
elif cmd_type == 'create_sphere':
# SIMPLIFIED APPROACH: Just create the sphere without complex checks
try:
center_x = cmd_data.get('center_x', 0)
center_y = cmd_data.get('center_y', 0)
center_z = cmd_data.get('center_z', 0)
radius = cmd_data.get('radius', 5.0)
doc = Rhino.RhinoDoc.ActiveDoc
if not doc:
raise Exception("No active Rhino document")
# Create the sphere directly
center = Rhino.Geometry.Point3d(center_x, center_y, center_z)
sphere = Rhino.Geometry.Sphere(center, radius)
# Convert to a brep for better display
brep = sphere.ToBrep()
# Add to document, ignoring the return value
doc.Objects.AddBrep(brep)
# Force view update
doc.Views.Redraw()
result = {
'status': 'success',
'message': f'Sphere created at ({center_x}, {center_y}, {center_z}) with radius {radius}'
}
except Exception as e:
result = {
'status': 'error',
'message': f'Sphere creation error: {str(e)}',
'traceback': traceback.format_exc()
}
elif cmd_type == 'refresh_view':
try:
doc = Rhino.RhinoDoc.ActiveDoc
if doc:
doc.Views.Redraw()
result = {'status': 'success', 'message': 'Views refreshed'}
else:
result = {'status': 'error', 'message': 'No active document'}
except Exception as e:
result = {'status': 'error', 'message': f'Refresh error: {str(e)}'}
elif cmd_type == 'run_script':
script = cmd_data.get('script', '')
if script:
# Capture print output
old_stdout = sys.stdout
from io import StringIO
captured_output = StringIO()
sys.stdout = captured_output
try:
# Execute the script
exec(script)
result = {
'status': 'success',
'message': 'Script executed',
'data': {'output': captured_output.getvalue()}
}
except Exception as e:
result = {
'status': 'error',
'message': f'Script execution error: {str(e)}',
'data': {'traceback': traceback.format_exc()}
}
finally:
sys.stdout = old_stdout
else:
result = {'status': 'error', 'message': 'No script provided'}
# Send the result back using the custom encoder
conn.sendall(json.dumps(result, cls=RhinoEncoder).encode('utf-8'))
except json.JSONDecodeError:
conn.sendall(json.dumps({
'status': 'error',
'message': 'Invalid JSON format'
}).encode('utf-8'))
except Exception as e:
print(f"Error processing command: {str(e)}")
traceback.print_exc()
conn.sendall(json.dumps({
'status': 'error',
'message': f'Server error: {str(e)}',
'traceback': traceback.format_exc()
}, cls=RhinoEncoder).encode('utf-8'))
except Exception as e:
print(f"Connection error: {str(e)}")
finally:
print(f"Connection closed with {addr}")
conn.close()
def start_server():
"""Start the socket server"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
s.bind((HOST, PORT))
s.listen()
print(f"Server started on {HOST}:{PORT}")
print("Waiting for connections...")
try:
while True:
conn, addr = s.accept()
client_thread = threading.Thread(target=handle_client, args=(conn, addr))
client_thread.daemon = True
client_thread.start()
except KeyboardInterrupt:
print("Server shutting down...")
except Exception as e:
print(f"Server error: {str(e)}")
traceback.print_exc()
except Exception as e:
print(f"Failed to bind to {HOST}:{PORT}. Error: {str(e)}")
print("Try closing any other running instances of this script or check if another program is using this port.")
# Display server information
print("\n========== RHINO BRIDGE ==========")
print(f"Version: {SERVER_VERSION}")
print(f"Started: {SERVER_START_TIME}")
print(f"File: {__file__}")
print("===================================\n")
# Start the server in a background thread to keep Rhino responsive
server_thread = threading.Thread(target=start_server)
server_thread.daemon = True
server_thread.start()
print("Rhino Bridge started!")
print(f"Listening on {HOST}:{PORT}")
print("Keep this script running in Rhino's Python editor")
print("The server will run until you close this script or Rhino")
```
### 2. Python Client Script (`rhino_client_deluxe.py`)
```python
"""
Rhino Client Deluxe - Interactive client for Rhino connection.
Version: 1.0 (2025-03-13)
This script provides an interactive terminal for connecting to and
controlling Rhino from external Python scripts.
"""
import os
import sys
import json
import socket
import time
from typing import Dict, Any, Optional, List, Tuple
class RhinoClient:
"""Client for maintaining an interactive connection with Rhino."""
def __init__(self, host: str = '127.0.0.1', port: int = 8888):
"""Initialize the interactive Rhino client."""
self.host = host
self.port = port
self.socket = None
self.connected = False
self.command_history: List[str] = []
def connect(self) -> bool:
"""Establish connection to the Rhino socket server."""
try:
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.connect((self.host, self.port))
self.connected = True
# Test connection with ping and verify server version
response = self.send_command('ping')
if response and response.get('status') == 'success':
server_data = response.get('data', {})
print(f"\n✅ Connected to Rhino {server_data.get('version', 'unknown')}")
print(f"🔌 Server: {server_data.get('server_version', 'unknown')}")
print(f"📂 Script: {server_data.get('script_path', 'unknown')}")
print(f"⏰ Started: {server_data.get('server_start_time', 'unknown')}")
# Ensure we're connected to the Deluxe server
if 'Deluxe' not in server_data.get('server_version', ''):
print("\n⚠️ WARNING: Not connected to RhinoServerDeluxe!")
print("You may experience errors with spheres and other commands.")
print("Please run rhino_server_deluxe.py in Rhino's Python editor.")
# Add an immediate view refresh to ensure everything is visible
self.refresh_view()
return True
print("\n❌ Connection test failed")
self.disconnect()
return False
except Exception as e:
print(f"\n❌ Connection error: {str(e)}")
return False
def disconnect(self) -> None:
"""Close the connection to Rhino."""
if self.socket:
try:
self.socket.close()
except:
pass
self.socket = None
self.connected = False
print("\nDisconnected from Rhino")
def send_command(self, cmd_type: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Send a command to the Rhino server and return the response."""
if not self.connected or not self.socket:
print("❌ Not connected to Rhino")
return {'status': 'error', 'message': 'Not connected to Rhino'}
try:
# Create command
command = {
'type': cmd_type,
'data': data or {}
}
# Send command
self.socket.sendall(json.dumps(command).encode('utf-8'))
# Receive response
response = self.socket.recv(8192)
return json.loads(response.decode('utf-8'))
except Exception as e:
print(f"❌ Error sending command: {str(e)}")
# Don't disconnect automatically to allow for retry
return {'status': 'error', 'message': f'Command error: {str(e)}'}
def create_sphere(self, x: float, y: float, z: float, radius: float) -> Dict[str, Any]:
"""Create a sphere in Rhino."""
data = {
'center_x': x,
'center_y': y,
'center_z': z,
'radius': radius
}
result = self.send_command('create_sphere', data)
# Always refresh view after creating geometry
if result.get('status') == 'success':
self.refresh_view()
return result
def run_script(self, script: str) -> Dict[str, Any]:
"""Run a Python script in Rhino."""
result = self.send_command('run_script', {'script': script})
# Always refresh view after running a script
if result.get('status') == 'success':
self.refresh_view()
return result
def refresh_view(self) -> Dict[str, Any]:
"""Refresh the Rhino viewport."""
return self.send_command('refresh_view')
def add_to_history(self, command: str) -> None:
"""Add a command to the history."""
if command and command not in ['', 'exit', 'help']:
self.command_history.append(command)
if len(self.command_history) > 100: # Limit history size
self.command_history.pop(0)
def print_help() -> None:
"""Print help information about available commands."""
print("\n=== Available Commands ===")
print("sphere <x> <y> <z> <radius> - Create a sphere")
print("script <python_code> - Run Python code in Rhino")
print("refresh - Refresh the Rhino viewport")
print("history - Show command history")
print("ping - Test connection to Rhino")
print("help - Show this help message")
print("exit - Close the connection and exit")
print("\nExample: sphere 10 20 0 5")
print("Example: script import Rhino; print(f\"Current doc: {Rhino.RhinoDoc.ActiveDoc.Name}\")")
def main() -> None:
"""Run the interactive Rhino client."""
print("\n========== RHINO CLIENT DELUXE ==========")
print("Version: 1.0 (2025-03-13)")
print("==========================================\n")
print("This script provides an interactive connection to Rhino.")
print("Make sure 'rhino_server_deluxe.py' is running in Rhino's Python editor.")
# Create client
client = RhinoClient()
# Connect to Rhino
if not client.connect():
print("\nFailed to connect to Rhino. Make sure the server script is running.")
return
print_help()
# Command loop
try:
while True:
command = input("\nrhino> ").strip()
if not command:
continue
if command.lower() == 'exit':
break
if command.lower() == 'help':
print_help()
continue
if command.lower() == 'ping':
response = client.send_command('ping')
if response.get('status') == 'success':
print(f"Connection active! Rhino version: {response.get('data', {}).get('version', 'unknown')}")
else:
print(f"Ping failed: {response.get('message', 'Unknown error')}")
client.add_to_history(command)
continue
# View refresh command
if command.lower() == 'refresh':
response = client.refresh_view()
if response.get('status') == 'success':
print("✅ Viewport refreshed")
else:
print(f"❌ Refresh failed: {response.get('message', 'Unknown error')}")
client.add_to_history(command)
continue
# Command history
if command.lower() == 'history':
if client.command_history:
print("\n=== Command History ===")
for i, cmd in enumerate(client.command_history):
print(f"{i+1}. {cmd}")
else:
print("No command history yet.")
continue
# Parse sphere command: sphere <x> <y> <z> <radius>
if command.lower().startswith('sphere '):
try:
parts = command.split()
if len(parts) != 5:
print("❌ Invalid sphere command. Format: sphere <x> <y> <z> <radius>")
continue
x, y, z, radius = map(float, parts[1:])
response = client.create_sphere(x, y, z, radius)
if response.get('status') == 'success':
print(f"✅ {response.get('message', 'Sphere created successfully')}")
else:
print(f"❌ Sphere creation failed: {response.get('message', 'Unknown error')}")
if response.get('traceback'):
print("\nError details:")
print(response.get('traceback'))
print("\nTroubleshooting tip: Make sure you're running rhino_server_deluxe.py in Rhino")
client.add_to_history(command)
except ValueError:
print("❌ Invalid parameters. All values must be numbers.")
continue
# Handle script command: script <python code>
if command.lower().startswith('script '):
script = command[7:] # Remove 'script ' prefix
response = client.run_script(script)
if response.get('status') == 'success':
print("\n=== Script Output ===")
output = response.get('data', {}).get('output', 'No output')
if output.strip():
print(output)
else:
print("Script executed successfully (no output)")
else:
print(f"❌ Script error: {response.get('message', 'Unknown error')}")
if 'traceback' in response:
print("\n=== Error Traceback ===")
print(response['traceback'])
client.add_to_history(command)
continue
print(f"❌ Unknown command: {command}")
print("Type 'help' for available commands")
except KeyboardInterrupt:
print("\nInterrupted by user")
except Exception as e:
print(f"\n❌ Error: {str(e)}")
import traceback
traceback.print_exc()
finally:
client.disconnect()
if __name__ == "__main__":
main()
```
### 3. Rhino Command Script (`PythonBridgeCommand.py`)
```python
"""Rhino-Python Bridge Command
Creates a custom Rhino command to start the Python socket server.
"""
import Rhino
import rhinoscriptsyntax as rs
import scriptcontext as sc
import System
import os
__commandname__ = "PythonBridge"
def RunCommand(is_interactive):
"""Run the PythonBridge command, which starts the socket server for Python connections."""
# Create path to the user's Documents folder (reliable location)
docs_folder = System.Environment.GetFolderPath(System.Environment.SpecialFolder.MyDocuments)
projects_folder = os.path.join(docs_folder, "CascadeProjects", "rhino_windsurf")
# Path to the server script
script_path = os.path.join(projects_folder, "rhino_bridge.py")
# Check if the file exists at the default location
if not os.path.exists(script_path):
# If not found, ask the user for the file location
filter = "Python Files (*.py)|*.py|All Files (*.*)|*.*||"
script_path = rs.OpenFileName("Select the rhino_bridge.py script", filter)
if not script_path:
print("Operation canceled. No server script selected.")
return Rhino.Commands.Result.Cancel
# Run the script using Rhino's script runner
Rhino.RhinoApp.RunScript("_-RunPythonScript \"" + script_path + "\"", False)
print("Python Bridge server started!")
print(f"Server script: {script_path}")
print("You can now connect from external Python scripts")
return Rhino.Commands.Result.Success
# This is needed for Rhino to recognize the command
if __name__ == "__main__":
RunCommand(True)
```
## Using the Python-Rhino Bridge
### Basic Commands
Once the bridge is set up and both the server and client are running, you can interact with Rhino using these commands:
1. **Creating Spheres**:
```
rhino> sphere <x> <y> <z> <radius>
```
Example: `sphere 10 20 0 5`
2. **Running Python Scripts in Rhino**:
```
rhino> script <python_code>
```
Example: `script import Rhino; print(f"Rhino version: {Rhino.RhinoApp.Version}")`
3. **Refreshing the Viewport**:
```
rhino> refresh
```
4. **Testing Connection**:
```
rhino> ping
```
5. **Viewing Command History**:
```
rhino> history
```
6. **Exiting the Client**:
```
rhino> exit
```
### Advanced Usage: Multi-Statement Python Commands
You can run multiple Python statements in a single command by separating them with semicolons:
```
rhino> script import Rhino; import random; x = random.uniform(0,10); y = random.uniform(0,10); print(f"Random point: ({x}, {y})")
```
For more complex scripts, you can use Python's block syntax with proper indentation:
```
rhino> script if True:
import Rhino
import random
for i in range(3):
x = random.uniform(0, 10)
y = random.uniform(0, 10)
z = random.uniform(0, 10)
print(f"Point {i+1}: ({x}, {y}, {z})")
```
### Creating Complex Geometry
To create more complex geometry, you can write scripts that leverage the full power of RhinoCommon:
```
rhino> script import Rhino; doc = Rhino.RhinoDoc.ActiveDoc; curve = Rhino.Geometry.Circle(Rhino.Geometry.Plane.WorldXY, 10).ToNurbsCurve(); doc.Objects.AddCurve(curve); doc.Views.Redraw()
```
## Troubleshooting
### Common Issues and Solutions
1. **Connection Refused**:
- Ensure Rhino is running with the server script active
- Check if another instance of the server is already running on port 8888
- Try restarting Rhino and the server script
2. **Geometry Not Appearing**:
- Use the `refresh` command to force a viewport update
- Make sure you have an active document open in Rhino
- Check for error messages in the command response
3. **Script Execution Errors**:
- Verify your Python syntax is correct
- Make sure you're using RhinoCommon API methods correctly
- Check the error traceback for specific issues
4. **Server Warning in Client**:
- The "Not connected to RhinoServerDeluxe" warning can be ignored if using the Bridge version
- Make sure you're using compatible versions of the server and client scripts
### Extending the Bridge
You can extend the functionality of the bridge by:
1. Adding new command types to the server script
2. Implementing additional geometry creation methods in the client
3. Creating specialized scriptlets for common tasks
## Conclusion
The Python-Rhino Bridge provides a powerful way to control Rhino programmatically from external Python scripts. Whether you're automating modeling tasks, integrating with other tools, or building custom workflows, this bridge offers a flexible and robust connection between Python and Rhino.
```