graph 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

pip install dcc-mcp-ipc

pip install "dcc-mcp-ipc[zeroconf]"

poetry add dcc-mcp-ipc

from dcc_mcp_ipc.server import create_dcc_server, DCCRPyCService


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)

from dcc_mcp_ipc.client import BaseDCCClient


client = BaseDCCClient("maya", host="localhost", port=18812)
client.connect()
result = client.call("get_scene_info")
client.disconnect()

from dcc_mcp_ipc.action_adapter import ActionAdapter, get_action_adapter


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, ...}

my_skills/
  create_light/
    SKILL.md      # frontmatter: name, description, tools, scripts
    run.py        # executed when the tool is called

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)

from dcc_mcp_ipc.client import ConnectionPool


pool = ConnectionPool()

with pool.get_client("maya", host="localhost") as client:
    result = client.call("execute_cmd", "sphere", radius=5)
    print(result)

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()

import os
from dcc_mcp_core import TransportAddress
from dcc_mcp_ipc.transport.ipc_transport import (
    IpcClientTransport,
    IpcServerTransport,
    IpcTransportConfig,
)

# Client side
config = IpcTransportConfig(host="localhost", port=19000)
transport = IpcClientTransport(config)
transport.connect()
result = transport.execute("get_scene_info")
transport.disconnect()

# Server side (inside DCC plugin)
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()

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};")

import asyncio
from dcc_mcp_ipc.client.async_dcc import AsyncDCCClient


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())

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