loading…
Search for a command to run...
loading…
Multi-protocol IPC adapter layer for integrating DCC software (Maya, Houdini, Blender, Unreal, Unity) with MCP, enabling tool registration and remote control ac
Multi-protocol IPC adapter layer for integrating DCC software (Maya, Houdini, Blender, Unreal, Unity) with MCP, enabling tool registration and remote control across multiple transport protocols.
Multi-protocol IPC adapter layer for DCC software integration with Model Context Protocol (MCP).
Built on top of dcc-mcp-core (Rust/PyO3 backend), it provides a high-performance, type-safe framework for exposing DCC functionality as MCP tools across multiple transport protocols.
Documentation: docs site | v2.0.0 (Unreleased) — Breaking changes ahead, see CHANGELOG.md
| Feature | Description |
|---|---|
| Protocol-agnostic | RPyC for embedded-Python DCCs (Maya/Houdini/Blender), HTTP for Unreal/Unity, WebSocket, and Rust-native IPC for maximum throughput. |
| Zero-code Skills | Drop a SKILL.md file into a directory — SkillManager auto-registers it as an MCP tool. No Python boilerplate needed. |
| Rust-powered core | Action dispatch, validation, and telemetry handled by dcc-mcp-core via PyO3; Python layer focuses on DCC-specific glue code. |
| Hot-reload skills | SkillWatcher monitors skill directories and re-registers tools on file changes without restarting the DCC. |
| Service discovery | ZeroConf (mDNS) + file-based fallback for automatic DCC server detection. |
| Connection pooling | ConnectionPool with auto-discovery for efficient client-side connection reuse. |
SKILL.md frontmatter with hot-reloadActionRegistry + ActionDispatcher (Rust) — JSON-serialised parameter dispatchrpyc, http, websocket, ipc)ServiceDiscoveryFactoryasyncio) for non-blocking operationsDCCAdapter) and servicesgraph TD
A[AI Assistant / MCP Client] --> B[MCP Server Layer]
B --> C[ActionAdapter\nActionRegistry + ActionDispatcher]
C --> D[RPyC Transport\nDCC Python env]
C --> E[IPC Transport\nRust FramedChannel]
C --> F[HTTP Transport\nUnreal / Unity REST]
C --> G[WebSocket Transport]
G[SkillManager\nSKILL.md auto-discovery] --> C
H[dcc-mcp-core\nRust/PyO3 backend] --> C
D --> I[DCC App\nMaya / Houdini / Blender]
E --> I
F --> J[DCC App\nUnreal / Unity]
style C fill:#e1f5fe
style H fill:#fff3e0
style G fill:#e8f5e9
| Component | Module | Description |
|---|---|---|
ActionAdapter |
action_adapter.py |
Wraps Rust ActionRegistry + ActionDispatcher; registers handlers and dispatches JSON-parameterised calls |
SkillManager |
skills/scanner.py |
Scans directories for SKILL.md skills, registers them as action handlers, supports hot-reload |
DCCServer |
server/dcc.py |
Manages the RPyC/IPC server lifecycle inside a DCC process |
BaseDCCClient |
client/base.py |
Core client connection/call logic with auto-discovery |
ConnectionPool |
client/pool.py |
Connection pooling for efficient resource management |
IpcClientTransport / IpcServerTransport |
transport/ipc_transport.py |
Rust-native framed-channel IPC, registered as "ipc" protocol |
ServiceDiscoveryFactory |
discovery/factory.py |
Strategy-pattern selector for ZeroConf or file-based discovery |
MockDCCService |
testing/mock_services.py |
Simulates DCC applications for testing |
pip install dcc-mcp-ipc
With optional ZeroConf support:
pip install "dcc-mcp-ipc[zeroconf]"
Or with Poetry:
poetry add dcc-mcp-ipc
class MayaService(DCCRPyCService): def get_scene_info(self): return {"scene": "Maya scene info"}
def exposed_execute_cmd(self, cmd_name, *args, **kwargs):
pass
server = create_dcc_server( dcc_name="maya", service_class=MayaService, port=18812, ) server.start(threaded=True)
</augment_code_snippet>
### Client-side
<augment_code_snippet path="README.md" mode="EXCERPT">
````python
from dcc_mcp_ipc.client import BaseDCCClient
client = BaseDCCClient("maya", host="localhost", port=18812)
client.connect()
result = client.call("get_scene_info")
client.disconnect()
The Action system is built on the Rust-backed ActionRegistry + ActionDispatcher. All parameters are JSON-serialised:
adapter = get_action_adapter("maya")
def create_sphere(radius: float = 1.0, name: str = "sphere1") -> dict: return {"success": True, "message": f"Created {name}", "context": {"name": name}}
adapter.register_action( "create_sphere", create_sphere, description="Create a sphere primitive", category="modeling", tags=["primitive", "mesh"], )
result = adapter.call_action("create_sphere", radius=2.0, name="mySphere") print(result.success) # True print(result.to_dict()) # {"success": True, ...}
</augment_code_snippet>
### Zero-code Skills via SkillManager
Drop a `SKILL.md` file into a directory structure:
```
my_skills/
create_light/
SKILL.md # frontmatter: name, description, tools, scripts
run.py # executed when the tool is called
```
<augment_code_snippet path="README.md" mode="EXCERPT">
````python
from dcc_mcp_ipc.skills import SkillManager
from dcc_mcp_ipc.action_adapter import get_action_adapter
adapter = get_action_adapter("maya")
mgr = SkillManager(adapter=adapter, dcc_name="maya")
mgr.load_paths(["/pipeline/skills"])
mgr.start_watching()
# Now "create_light" is callable as an MCP tool
result = adapter.call_action("create_light", intensity=100.0)
pool = ConnectionPool()
with pool.get_client("maya", host="localhost") as client: result = client.call("execute_cmd", "sphere", radius=5) print(result)
</augment_code_snippet>
### Service Factories
Three factory patterns are available for different lifecycle needs:
<augment_code_snippet path="README.md" mode="EXCERPT">
````python
from dcc_mcp_ipc.server import (
create_service_factory,
create_shared_service_instance,
create_raw_threaded_server,
)
class SceneManager:
def __init__(self):
self.scenes = {}
def add_scene(self, name, data):
self.scenes[name] = data
scene_manager = SceneManager()
# Per-connection instance
service_factory = create_service_factory(MayaService, scene_manager)
# Shared singleton instance
shared_service = create_shared_service_instance(MayaService, scene_manager)
# Raw threaded server
server = create_raw_threaded_server(service_factory, port=18812)
server.start()
Zero-copy low-latency messaging via the Rust core:
config = IpcTransportConfig(host="localhost", port=19000) transport = IpcClientTransport(config) transport.connect() result = transport.execute("get_scene_info") transport.disconnect()
def handle_channel(channel): msg = channel.recv() # process and respond ...
addr = TransportAddress.default_local("maya", os.getpid()) server = IpcServerTransport(addr, handler=handle_channel) bound_addr = server.start()
</augment_code_snippet>
### Creating a Custom DCC Adapter
<augment_code_snippet path="README.md" mode="EXCERPT">
````python
from dcc_mcp_ipc.adapter import DCCAdapter
from dcc_mcp_ipc.client import BaseDCCClient
class MayaAdapter(DCCAdapter):
def _initialize_client(self) -> None:
self.client = BaseDCCClient(
dcc_name="maya",
host=self.host,
port=self.port,
connection_timeout=self.connection_timeout,
)
def create_sphere(self, radius: float = 1.0):
self.ensure_connected()
assert self.client is not None
return self.client.execute_dcc_command(f"sphere -r {radius};")
async def main(): client = AsyncDCCClient("maya", host="localhost", port=18812) await client.connect() result = await client.call("get_scene_info") await client.disconnect()
asyncio.run(main())
</augment_code_snippet>
### Testing with Mock Services
<augment_code_snippet path="README.md" mode="EXCERPT">
````python
import threading
from dcc_mcp_ipc.testing.mock_services import MockDCCService
from dcc_mcp_ipc.client import BaseDCCClient
server = MockDCCService.start(port=18812)
client = BaseDCCClient("mock_dcc", host="localhost", port=18812)
client.connect()
info = client.get_dcc_info()
print(info) # {"name": "mock_dcc", ...}
client.disconnect()
server.stop()
git clone https://github.com/loonghao/dcc-mcp-ipc.git
cd dcc-mcp-ipc
poetry install
nox -s pytest # Run tests
nox -s lint # Lint (mypy + ruff + isort)
nox -s lint-fix # Auto-fix lint issues
nox -s build # Build distribution packages
dcc-mcp-ipc/
├── src/dcc_mcp_ipc/
│ ├── __init__.py # Lazy-import public API surface
│ ├── action_adapter.py # Action system (Rust-backed)
│ ├── adapter/ # DCC & application adapters
│ ├── client/ # Synchronous & async clients + pool
│ ├── server/ # RPyC server + factories + lifecycle
│ ├── transport/ # RPyC / HTTP / WS / IPC transports
│ ├── discovery/ # ZeroConf + file-based service discovery
│ ├── skills/ # SkillManager zero-code system
│ ├── scene/ # Scene operations interface
│ ├── snapshot/ # Snapshot interface
│ ├── application/ # Generic application adapter/service/client
│ ├── testing/ # Mock services for testing
│ └── utils/ # Errors, DI, decorators, RPyC helpers
├── tests/ # 68 test files mirroring source layout
├── examples/ # Usage examples
├── docs/ # VitePress documentation site
└── nox_actions/ # Nox task definitions
Выполни в терминале:
claude mcp add dcc-mcp-ipc -- npx Безопасность
Низкий рискАвтоматическая эвристика по публичным данным — не гарантия безопасности.