# dcc-mcp-core — Full API Reference

> This document provides comprehensive API documentation for AI agents and downstream projects
> that depend on dcc-mcp-core. For a concise overview, see [llms.txt](llms.txt).
>
> **For AI agents**: Start with the [Quick Decision Guide](#quick-decision-guide) to pick the right API for your task.

---

## Quick Decision Guide

| I want to… | Use this API |
|-------------|--------------|
| Register a script/tool for AI to call | `ToolRegistry.register()` |
| Return a result from a DCC action | `success_result()` / `error_result()` |
| Auto-discover skill packages from directories | `scan_and_load()` / `SkillScanner` |
| Watch a skills directory for live updates | `SkillWatcher` |
| Validate action input parameters | `ToolValidator` / `InputValidator` |
| Route action calls to Python handlers | `ToolDispatcher` |
| Define an MCP Tool for the LLM | `ToolDefinition` + `ToolAnnotations` |
| Connect to a running DCC process | `connect_ipc(TransportAddress)` |
| Launch a new DCC process | `PyDccLauncher` |
| Monitor DCC process health | `PyProcessWatcher` / `PyProcessMonitor` |
| Enforce security policy on actions | `SandboxPolicy` + `SandboxContext` |
| Track performance metrics | `ToolRecorder` + `ToolMetrics` |
| Share large data between processes | `PySharedSceneBuffer` (LZ4 compressed) |
| Capture a DCC viewport screenshot | `Capturer.new_auto().capture()` (full-screen) / `Capturer.new_window_auto().capture_window(process_id=pid)` (single window) |
| Bind diagnostics MCP tools to a specific DCC instance | `DccServerBase(..., dcc_pid=pid, dcc_window_title=title)` |
| Expose a group of tools only when asked | Declare `groups:` in SKILL.md + `activate_tool_group` / `deactivate_tool_group` |
| Exchange scene data via USD format | `UsdStage` + `scene_info_json_to_stage()` |
| Bridge DCC protocols | `BridgeRegistry` + `register_bridge()` / `get_bridge_context()` |
| Describe scene hierarchy | `SceneNode` + `SceneObject` + `ObjectTransform` + `BoundingBox` |
| Serialize result for transport | `serialize_result()` / `deserialize_result()` with `SerializeFormat` |
| Wrap values for safe RPyC transport | `wrap_value()` / `unwrap_value()` |
| Handle multiple versions of an action | `VersionedRegistry` + `VersionConstraint` |
| Expose DCC tools over HTTP/MCP | `create_skill_server("maya", McpHttpConfig(port=8765))` — Skills-First one-call setup; falls back to `McpHttpServer(registry, config)` for manual registry wiring |

---

## Architecture

dcc-mcp-core is a **Rust workspace** with Python bindings via PyO3. All logic lives in Rust
sub-crates; the root crate re-exports everything into a single `dcc_mcp_core._core` Python
extension module. The Python package `dcc_mcp_core` re-exports all public APIs from `_core`.

**15 crates** — **Zero runtime Python dependencies** — install with just `pip install dcc-mcp-core`.

```
crates/
├── dcc-mcp-naming/       # SEP-986 tool-name / action-id validators (TOOL_NAME_RE, validate_tool_name)
├── dcc-mcp-models/       # ToolResult, SkillMetadata
├── dcc-mcp-actions/      # ToolRegistry, ToolDispatcher, ToolValidator, EventBus, ToolPipeline
├── dcc-mcp-skills/       # SkillScanner, SkillWatcher, parse_skill_md, dependency resolver
├── dcc-mcp-protocols/    # MCP type definitions (Tool, Resource, Prompt, DccAdapter)
├── dcc-mcp-transport/    # IPC transport, TransportManager, ConnectionPool, CircuitBreaker
├── dcc-mcp-http/         # McpHttpServer, McpHttpConfig, McpServerHandle (MCP Streamable HTTP 2025-03-26)
├── dcc-mcp-process/      # PyDccLauncher, PyProcessMonitor, PyProcessWatcher, PyCrashRecoveryPolicy
├── dcc-mcp-telemetry/    # TelemetryConfig, RecordingGuard, ToolMetrics, ToolRecorder
├── dcc-mcp-sandbox/      # SandboxPolicy, SandboxContext, InputValidator, AuditLog, AuditEntry
├── dcc-mcp-shm/          # PyBufferPool, PySharedBuffer, PySharedSceneBuffer, LZ4 compression
├── dcc-mcp-capture/      # Capturer, CaptureFrame, CaptureResult, CaptureTarget, WindowFinder (HWND on Windows)
├── dcc-mcp-usd/          # UsdStage, UsdPrim, SdfPath, VtValue, scene info bridge
├── dcc-mcp-server/       # Binary entry point, gateway runner
└── dcc-mcp-utils/        # Filesystem, constants, type wrappers, JSON conversion
```

**Key import pattern** (always use the top-level package):
```python
from dcc_mcp_core import ToolRegistry, success_result, SkillScanner
# Never: from dcc_mcp_core._core import ...
```

---

## Models (`dcc_mcp_core`)

### ToolResult

Structured result for all action executions. AI-friendly: the `prompt` field carries next-step
hints for the LLM. Immutable snapshot — use `with_error()`/`with_context()` to derive.

**Constructor:**
```python
ToolResult(success=True, message="", prompt=None, error=None, context=None)
```

**Properties (read-only except `message`):**
- `success: bool` — Whether the execution was successful
- `message: str` — Human-readable result description (settable)
- `prompt: Optional[str]` — Suggestion for AI about next steps ← **use this for AI guidance**
- `error: Optional[str]` — Error message when success is False
- `context: dict` — Additional context data (returns a new dict each access)

**Methods:**
- `with_error(error: str) -> ToolResult` — Copy with error info (sets success=False)
- `with_context(**kwargs) -> ToolResult` — Copy with updated context
- `to_dict() -> dict` — Full dict representation
- `to_json() -> str` — JSON string representation
- `keys() -> list[str]` — Dict keys (mapping protocol)
- `__iter__()` — Iterate over keys (mapping protocol)

**Factory functions (preferred for creation):**
```python
from dcc_mcp_core import success_result, error_result, from_exception, validate_action_result

# success_result: positional message, optional prompt, remaining kwargs → context
result = success_result("Created 5 spheres", prompt="Use modify_spheres next", count=5)
# result.context == {"count": 5}

# error_result: message, error string, optional prompt+solutions
error = error_result("Failed", "File not found", prompt="Check path", possible_solutions=["fix A", "fix B"])

# from_exception: wrap an exception string (not the exception object itself)
try:
    risky_op()
except Exception as e:
    exc_result = from_exception(str(e), message="Import failed", include_traceback=True)

# validate_action_result: normalize any type to ToolResult
validated = validate_action_result({"success": True, "message": "ok"})  # dict → ToolResult
validated = validate_action_result("hello")  # wraps non-dict as success with context["value"]
validated = validate_action_result(None)     # returns a success result
```

### SkillMetadata

Metadata parsed from SKILL.md frontmatter. All fields are get/set.

**Fields:**
- `name: str` — Unique identifier (used in action naming)
- `description: str` — Human-readable description (default: "")
- `tools: List[ToolDeclaration]` — MCP tool declarations (name, source_file, schemas); default: []
- `allowed_tools: List[str]` — Agent permission list, e.g. ["Bash", "Read"] (agentskills.io spec); default: []
- `license: str` — License identifier, e.g. "MIT" (agentskills.io spec); default: ""
- `compatibility: str` — Runtime compatibility string, e.g. "Python>=3.9" (agentskills.io spec); default: ""
- `dcc: str` — Target DCC application (default: "python")
- `tags: List[str]` — Classification tags (default: [])
- `scripts: List[str]` — Discovered script file paths (populated by loader; absolute paths)
- `skill_path: str` — Absolute path to skill directory (populated by loader)
- `version: str` — Skill version (default: "1.0.0")
- `depends: List[str]` — Names of other skills this skill requires (default: [])
- `metadata_files: List[str]` — Files discovered under metadata/ directory (default: [])
- `external_deps: Optional[str]` — External dependency declaration as JSON string (MCP servers, env vars, binaries). Set via `md.external_deps = json.dumps(deps)`, read via `json.loads(md.external_deps)`. `None` when not set. See `docs/guide/skill-scopes-policies.md` for the full schema.

**Action naming from SkillMetadata:**
```python
# Given: skill.name = "maya-geometry", script = "create_sphere.py"
# Action name = "maya_geometry__create_sphere"  (hyphens→underscores, double __ separator)
```

---

## Actions (`dcc_mcp_core`)

### ToolRegistry

Thread-safe tool registry using DashMap (Rust). Each instance is independent (no singleton).

```python
reg = dcc_mcp_core.ToolRegistry()

# Register — all params except name are optional
reg.register(
    name="create_sphere",
    description="Create a sphere",
    category="geometry",
    tags=["geo", "create"],
    dcc="maya",
    version="1.0.0",
    input_schema='{"type": "object", "required": ["radius"], "properties": {"radius": {"type": "number"}}}',
    output_schema='{"type": "object"}',
    source_file="/path/to/action.py",
)

# Query
meta = reg.get_action("create_sphere")                    # -> dict or None
meta = reg.get_action("create_sphere", dcc_name="maya")   # DCC-scoped lookup (preferred)
names = reg.list_actions_for_dcc("maya")                   # -> List[str]
all_actions = reg.list_actions()                           # -> List[dict]
all_actions = reg.list_actions(dcc_name="maya")            # -> List[dict] scoped
dccs = reg.get_all_dccs()                                  # -> List[str]

# Batch registration (v0.12.6+)
# More efficient than calling register() in a loop; silently skips entries without "name".
reg.register_batch([
    {"name": "create_sphere", "category": "geometry", "tags": ["create"], "dcc": "maya"},
    {"name": "delete_mesh",   "category": "edit",     "tags": ["delete"], "dcc": "maya"},
    {"name": "render_scene",  "category": "render",                       "dcc": "maya"},
])

# Unregister (v0.12.6+)
# Global: removes from the global registry AND all per-DCC maps.
removed = reg.unregister("create_sphere")                  # -> bool (True if found)
# Scoped: removes only from the specified DCC's map.
# The global entry is cleared only when no other DCC still references the action.
removed = reg.unregister("create_sphere", dcc_name="maya") # -> bool

# Search & discovery (v0.12.5+)
# search_actions: all filters are AND-ed; None / [] = no filter
geo_actions = reg.search_actions(category="geometry")               # by category
mesh_actions = reg.search_actions(tags=["mesh"])                    # by tag
create_geo  = reg.search_actions(category="geometry", tags=["create"], dcc_name="maya")
categories = reg.get_categories()                                   # sorted unique categories
tags       = reg.get_tags(dcc_name="maya")                          # sorted unique tags in maya

# Manage
reg.reset()    # clear all entries
len(reg)       # count of registered actions
repr(reg)      # "ToolRegistry(actions=N)"
```

**Returned dict structure from `get_action()` / `list_actions()`:**
```python
{
    "name": "create_sphere",
    "description": "Create a sphere",
    "category": "geometry",
    "tags": ["geo", "create"],
    "dcc": "maya",
    "version": "1.0.0",
    "input_schema": "{...}",   # JSON string
    "output_schema": "{...}",  # JSON string (empty if not set)
    "source_file": "/path/to/action.py" | None,
}
```

### ToolValidator

Validates JSON-encoded action parameters against a JSON Schema.

```python
import json
from dcc_mcp_core import ToolRegistry, ToolValidator

schema = json.dumps({
    "type": "object",
    "required": ["radius"],
    "properties": {"radius": {"type": "number", "minimum": 0.0}}
})
v = ToolValidator.from_schema_json(schema)
ok, errors = v.validate('{"radius": 1.0}')   # -> (True, [])
ok, errors = v.validate("{}")                  # -> (False, ["...missing required: radius"])

# Or build from registry:
v2 = ToolValidator.from_action_registry(reg, "create_sphere", dcc_name="maya")
```

### ToolDispatcher

Routes action calls to registered Python callables with automatic schema validation.

```python
import json
from dcc_mcp_core import ToolRegistry, ToolDispatcher

reg = ToolRegistry()
reg.register("create_sphere",
    input_schema=json.dumps({"type": "object", "required": ["radius"],
                              "properties": {"radius": {"type": "number", "minimum": 0.0}}}))

dispatcher = ToolDispatcher(reg)

def create_sphere_handler(params: dict):
    return {"created": True, "radius": params["radius"]}

dispatcher.register_handler("create_sphere", create_sphere_handler)
# Dispatch: validates params, calls handler, returns result dict
result = dispatcher.dispatch("create_sphere", json.dumps({"radius": 2.0}))
# result == {"action": "create_sphere", "output": {"created": True, "radius": 2.0},
#            "validation_skipped": False}

# Control validation behaviour
dispatcher.skip_empty_schema_validation = True  # skip validation if schema is "{}"

# Other methods
dispatcher.has_handler("create_sphere")  # -> bool
dispatcher.handler_count()               # -> int
dispatcher.handler_names()               # -> List[str] sorted
dispatcher.remove_handler("create_sphere")  # -> bool
```

### ToolPipeline

Middleware-style processing pipeline for actions. Wraps `ToolDispatcher` with composable
cross-cutting concerns (logging, timing, auditing, rate limiting, custom Python hooks).

Middleware runs in registration order for `before_dispatch`, in **reverse** for `after_dispatch`
(standard onion model).

```python
from dcc_mcp_core import (
    ToolRegistry, ToolDispatcher, ToolPipeline,
    LoggingMiddleware, TimingMiddleware, AuditMiddleware, RateLimitMiddleware,
)

# Build registry + dispatcher with handlers
reg = ToolRegistry()
reg.register("create_sphere", description="Create sphere", category="geometry")
dispatcher = ToolDispatcher(reg)
dispatcher.register_handler("create_sphere", lambda params: {"name": "sphere1"})

# Wrap in pipeline
pipeline = ToolPipeline(dispatcher)

# Add built-in middleware
pipeline.add_logging(log_params=True)             # tracing before/after
timing  = pipeline.add_timing()                   # returns TimingMiddleware handle
audit   = pipeline.add_audit(record_params=True)  # returns AuditMiddleware handle
rl      = pipeline.add_rate_limit(max_calls=10, window_ms=1000)  # RateLimitMiddleware handle

# Add Python callable hooks
pipeline.add_callable(
    before_fn=lambda action: print(f"before: {action}"),
    after_fn=lambda action, success: print(f"after: {action} ok={success}"),
)

# Register additional handlers directly on pipeline
pipeline.register_handler("delete_sphere", lambda params: True)

# Dispatch — returns dict with action/output/validation_skipped
result = pipeline.dispatch("create_sphere", '{"radius": 1.0}')
# result == {"action": "create_sphere", "output": {"name": "sphere1"}, "validation_skipped": bool}

# Query middleware state
timing.last_elapsed_ms("create_sphere")  # int | None (ms since last dispatch)
audit.records()                           # list[dict] — action/success/error/timestamp_ms
audit.records_for_action("create_sphere")
audit.record_count()
audit.clear()
rl.call_count("create_sphere")  # calls in current window
rl.max_calls                    # int
rl.window_ms                    # int

# Introspect pipeline
pipeline.middleware_count()   # int
pipeline.middleware_names()   # ["logging","timing","audit","rate_limit","python_callable"]
pipeline.handler_count()      # int

# Standalone middleware classes (can also be constructed independently)
lm = LoggingMiddleware(log_params=False)   # lm.log_params
tm = TimingMiddleware()                    # tm.last_elapsed_ms(action)
am = AuditMiddleware(record_params=True)   # am.records(), .clear()
rm = RateLimitMiddleware(max_calls=5, window_ms=500)  # rm.call_count(action)
```



### EventBus

Thread-safe publish/subscribe event system.

```python
bus = dcc_mcp_core.EventBus()

# Subscribe (returns subscriber ID for unsubscription)
sub_id = bus.subscribe("action_completed", lambda **kw: print(f"Done: {kw}"))
sub_id2 = bus.subscribe("action_completed", on_action_done)

# Publish with keyword arguments
bus.publish("action_completed", action="create_sphere", success=True, duration_ms=42)

# Unsubscribe by event + subscriber ID
removed = bus.unsubscribe("action_completed", sub_id)  # -> bool

repr(bus)  # "EventBus(subscriptions=N)"
```

### VersionedRegistry

Multi-version action registry allowing multiple versions of the same `(name, dcc)` pair.

```python
from dcc_mcp_core import VersionedRegistry, VersionConstraint, SemVer

vr = VersionedRegistry()
vr.register_versioned("create_sphere", "maya", "1.0.0", description="v1 basic")
vr.register_versioned("create_sphere", "maya", "1.5.0", description="v1.5 with subdivisions")
vr.register_versioned("create_sphere", "maya", "2.0.0", description="v2 GPU accelerated")

# Resolve best match (highest version satisfying constraint)
result = vr.resolve("create_sphere", "maya", "^1.0.0")
# result == {"name": "create_sphere", "dcc": "maya", "version": "1.5.0", ...}

# Resolve all matching
all_v1 = vr.resolve_all("create_sphere", "maya", "^1.0.0")
# -> [{"version": "1.0.0",...}, {"version": "1.5.0",...}]

# Query
vr.versions("create_sphere", "maya")         # -> ["1.0.0", "1.5.0", "2.0.0"]
vr.latest_version("create_sphere", "maya")   # -> "2.0.0"
vr.total_entries()                            # -> 3
vr.keys()                                     # -> [("create_sphere", "maya")]

# Remove (returns count of versions removed)
removed = vr.remove("create_sphere", "maya", "^1.0.0")   # removes 1.0.0 and 1.5.0 -> 2
removed = vr.remove("create_sphere", "maya", "*")         # removes all -> remaining count
```

### CompatibilityRouter

Returned by `VersionedRegistry.router()`. Resolves the best-matching tool version
given a client-side version constraint. **Not yet exported as a standalone Python class** —
access via `registry.router()`.

```python
from dcc_mcp_core import VersionedRegistry, VersionConstraint

vr = VersionedRegistry()
vr.register_versioned("create_sphere", "maya", "1.0.0")
vr.register_versioned("create_sphere", "maya", "2.0.0")

# Obtain a router for version-aware resolution
router = vr.router()
# The router borrows the registry and provides constraint-based lookups
# Use VersionedRegistry.resolve() as the primary API — the router is
# for advanced multi-step resolution scenarios
```

### SemVer

```python
from dcc_mcp_core import SemVer

v = SemVer(1, 2, 3)
str(v)              # "1.2.3"
v.major, v.minor, v.patch  # 1, 2, 3
SemVer.parse("2.0.0") > v  # True
SemVer.parse("v1.5.0-alpha")  # SemVer(1, 5, 0)
```

### VersionConstraint

```python
from dcc_mcp_core import VersionConstraint, SemVer

# Supported operators: * = >= > <= < ^ ~
c = VersionConstraint.parse("^1.0.0")   # same major, >= minor.patch
c.matches(SemVer(1, 5, 0))   # True
c.matches(SemVer(2, 0, 0))   # False

c2 = VersionConstraint.parse(">=1.2.0")
c3 = VersionConstraint.parse("~1.2.3")   # same major.minor, >= patch
c4 = VersionConstraint.parse("*")        # any version
```

---

## Skills System (`dcc_mcp_core`)

### SkillScanner

Discovers SKILL.md files in directories with mtime-based caching.

```python
scanner = dcc_mcp_core.SkillScanner()
dirs = scanner.scan(
    extra_paths=["/my/skills", "examples/skills"],
    dcc_name="maya",       # also checks platform-specific skills dir
    force_refresh=False,   # True to ignore mtime cache
)
# dirs: List[str] — absolute paths to skill directories containing SKILL.md

scanner.discovered_skills  # -> List[str] (last scan result, same as return value)
scanner.clear_cache()      # reset mtime cache — next scan will re-read all dirs
```

**Search path priority** (highest to lowest):
1. `extra_paths` argument
2. `DCC_MCP_SKILL_PATHS` environment variable (colon/semicolon separated)
3. Platform skills dir: `get_skills_dir(dcc_name)`

### SkillWatcher

Hot-reload watcher for skill directories. Monitors filesystem events using platform-native APIs
(inotify on Linux, FSEvents on macOS, ReadDirectoryChangesW on Windows).

```python
watcher = dcc_mcp_core.SkillWatcher(debounce_ms=300)  # coalesce rapid changes

watcher.watch("/path/to/skills")   # starts watching + does immediate reload
# watcher.watch() can be called multiple times for multiple directories

skills = watcher.skills()          # -> List[SkillMetadata] (immutable snapshot)
watcher.skill_count()              # -> int
watcher.watched_paths()            # -> List[str]
watcher.reload()                   # manual force reload

watcher.unwatch("/path/to/skills") # -> bool (True if was being watched)
```

### parse_skill_md

```python
meta = dcc_mcp_core.parse_skill_md("/path/to/skill-dir")  # -> SkillMetadata or None
```

Parses YAML frontmatter from SKILL.md, enumerates scripts/ directory (all supported extensions),
discovers metadata/ directory files, and merges `depends` from `metadata/depends.md`.

Returns `None` if the directory has no SKILL.md.

### scan_skill_paths

```python
dirs = dcc_mcp_core.scan_skill_paths(extra_paths=["/my/skills"], dcc_name="maya")
# Convenience: creates SkillScanner, scans, returns List[str] paths
```

### scan_and_load / scan_and_load_lenient

```python
# Full pipeline: scan → parse → resolve dependencies (topological sort)
skills, skipped = dcc_mcp_core.scan_and_load(
    extra_paths=["/my/skills"],
    dcc_name="maya",
)
# Returns (List[SkillMetadata], List[str] skipped_dirs)
# Raises ValueError on missing dependencies or cycles

# Lenient variant: skips unresolvable skills instead of raising
skills, skipped = dcc_mcp_core.scan_and_load_lenient(dcc_name="maya")
# Only raises on cyclic dependencies

for s in skills:
    print(f"{s.name} v{s.version}: {len(s.scripts)} scripts → {s.skill_path}")
    # Action names: f"{s.name.replace('-', '_')}__{stem}" for stem in script stems
```

### Dependency Resolution

```python
from dcc_mcp_core import resolve_dependencies, validate_dependencies, expand_transitive_dependencies

# Validate declared deps exist in a skill list
errors = validate_dependencies(skills)  # -> List[str] error messages; empty if OK

# Topological sort (raises if missing dep or cycle)
ordered = resolve_dependencies(skills)  # -> List[SkillMetadata] sorted

# Full transitive closure for a specific skill
all_deps = expand_transitive_dependencies(skills, "my-skill")  # -> List[str] names
```

### SKILL.md Format

```yaml
---
name: maya-geometry          # Required: unique identifier (kebab-case, max 64 chars)
description: "Maya geometry creation and modification tools. Use when creating or modifying polygon geometry."
allowed-tools: Bash(git:*) Read  # Space-separated pre-approved tools (agentskills.io spec, experimental)
tags: ["maya", "geometry"]   # Classification tags
dcc: maya                    # Target DCC: maya, blender, houdini, 3dsmax, python
version: "1.0.0"             # Semantic version
license: "MIT"               # License identifier (agentskills.io spec)
compatibility: "Maya 2024+"  # Environment requirements, max 500 chars (agentskills.io spec)
depends: ["other-skill"]     # Names of required skills (optional)
search-hint: "polygon modeling, sphere, bevel, extrude"  # dcc-mcp-core extension: keyword hints
---
# Human-readable description (markdown body)

This markdown body describes the skill to humans and AI agents.
Include: purpose, usage examples, expected environment, prerequisites.
```

**agentskills.io standard fields** (V1.0 spec, 2025-12-18): `name` (required), `description` (required), `license`, `compatibility`, `metadata`, `allowed-tools` (experimental — space-separated tool strings like `Bash(git:*) Read`).
**dcc-mcp-core extensions**: `dcc`, `tags`, `search-hint`, `tools`, `groups`, `depends`, `external_deps`, `next-tools`.

### Supported Script Extensions

`.py`, `.mel`, `.ms`, `.bat`, `.cmd`, `.sh`, `.bash`, `.ps1`, `.jsx`, `.js`

### Environment Variable

```bash
# Unix/macOS
export DCC_MCP_SKILL_PATHS="/path/to/skills1:/path/to/skills2"

# Windows
set DCC_MCP_SKILL_PATHS=C:\skills1;C:\skills2

# DCC-specific paths (set by DCC integration plugins or manually)
export DCC_MCP_MAYA_SKILL_PATHS="/maya/skills1:/maya/skills2"
export DCC_MCP_BLENDER_SKILL_PATHS="/blender/skills"
```

```python
# Read DCC-specific paths programmatically
from dcc_mcp_core import get_app_skill_paths_from_env
maya_paths = get_app_skill_paths_from_env("maya")    # reads DCC_MCP_MAYA_SKILL_PATHS
blender_paths = get_app_skill_paths_from_env("blender")
```

### SkillCatalog (Progressive Loading)

`SkillCatalog` wraps a `SkillScanner` with progressive skill loading and thread-safe state.

```python
from dcc_mcp_core import SkillCatalog, SkillScanner, SkillSummary

# Create catalog (preferred: use create_skill_server for Skills-First)
scanner = SkillScanner()
catalog = SkillCatalog(scanner)

# Discover available skills
catalog.discover(extra_paths=["/my/skills"], dcc_name="maya")

# List skills
all_skills: list[SkillSummary] = catalog.list_skills()
loaded_skills = catalog.list_skills(status="loaded")

# Search — matches name, description, search_hint, and tool names
results = catalog.find_skills(query="geometry", tags=["maya"], dcc="maya")

# Load / unload individual skills
catalog.load_skill("maya-geometry")      # -> bool
catalog.unload_skill("maya-geometry")    # -> bool
catalog.is_loaded("maya-geometry")       # -> bool
catalog.loaded_count()                   # -> int

# Get full metadata
meta = catalog.get_skill_info("maya-geometry")  # -> Optional[SkillMetadata]

# Tool groups (progressive exposure) — see next section
catalog.active_groups("maya-geometry")            # -> List[str]
catalog.activate_group("maya-geometry", "rigging")   # -> bool
catalog.deactivate_group("maya-geometry", "rigging") # -> bool
catalog.list_groups("maya-geometry")               # -> List[SkillGroup]
catalog.list_tools_catalog("maya-geometry")        # -> {group_name: [tool_names]}
```

**SkillSummary fields:** `.name`, `.description`, `.search_hint`, `.version`, `.dcc`, `.tags`, `.tool_count`, `.tool_names`, `.loaded`

### Tool Groups (Progressive Exposure)

Large skills can declare **tool groups** in SKILL.md so the AI client activates
only the toolsets it needs, keeping `tools/list` compact while all tools stay
discoverable via `search_tools`.

```yaml
# SKILL.md frontmatter
---
name: maya-geometry
dcc: maya
groups:
  - name: modeling
    description: "Polygon modeling and UV tools"
    default_active: true          # active at load_skill time
    tools: [create_sphere, extrude]
  - name: rigging
    description: "Skeleton, joints, skinning"
    default_active: false         # dormant until activate_tool_group
    tools: [create_joint]
tools:
  - name: create_sphere
    group: modeling
    source_file: scripts/create_sphere.py
  - name: create_joint
    group: rigging
    source_file: scripts/create_joint.py
---
```

**`SkillGroup` fields:** `.name`, `.description`, `.default_active` (bool), `.tools` (`List[str]`).

**Runtime behaviour after `load_skill`:**

1. Every tool is registered in `ToolRegistry` with its group metadata set.
2. Tools whose group has `default_active=false` are hidden from
   MCP `tools/list` — they remain in the registry and become visible once activated.
3. `tools/list` also emits `__group__<skill>.<group>` stubs for inactive
   groups so clients can discover and activate them on demand.

**Activate / deactivate from Python:**

```python
from dcc_mcp_core import ToolRegistry

# Via the registry (emits notifications/tools/list_changed)
registry.activate_tool_group("maya-geometry", "rigging")   # -> int enabled count
registry.deactivate_tool_group("maya-geometry", "rigging") # -> int disabled count
registry.list_tools_in_group("maya-geometry", "modeling")  # -> List[dict]
registry.list_actions_enabled()                            # -> List[dict]
registry.set_tool_enabled("maya_geometry__create_joint", True)
```

**Via MCP tools** — `create_skill_server` / `McpHttpServer` register three
group-control tools alongside the six skill-discovery tools:

| MCP tool | Description |
|----------|-------------|
| `activate_tool_group` | Enable all tools in a group; sends `notifications/tools/list_changed` |
| `deactivate_tool_group` | Disable all tools in a group |
| `search_tools` | Keyword search across currently-enabled tools |

### `next-tools` — Follow-Up Tool Hints (dcc-mcp-core extension)

The `next-tools` field guides AI agents to appropriate follow-up actions after a tool
executes. This is a dcc-mcp-core extension, not part of the agentskills.io specification.

```yaml
tools:
  - name: create_sphere
    description: "Create a polygon sphere"
    source_file: scripts/create_sphere.py
    next-tools:
      on-success: [maya_geometry__bevel_edges]      # suggest after success
      on-failure: [dcc_diagnostics__screenshot]      # debug on failure
```

| Key | Type | Description |
|-----|------|-------------|
| `on-success` | `List[str]` | Suggested tools after successful execution |
| `on-failure` | `List[str]` | Debugging/recovery tools on failure |

Both accept fully-qualified tool names in `{skill_name}__{tool_name}` format.

### Action Naming Convention

Each script in `scripts/` becomes an action:
```
{skill_name}__{script_stem}
```
- Hyphens in skill names → underscores
- Double underscore (`__`) separator
- Examples:
  - `maya-geometry/scripts/create_sphere.py` → `maya_geometry__create_sphere`
  - `git-tools/scripts/commit.sh` → `git_tools__commit`

---

## Skill Script Helpers — `dcc_mcp_core.skill` (pure Python)

A **pure-Python** sub-module for skill script authors. No compiled `_core` extension required — works inside any DCC Python environment.

```python
from dcc_mcp_core.skill import skill_entry, skill_success, skill_error, skill_warning, skill_exception, run_main
```

All helpers return a plain `dict` compatible with `ToolResult`.

### Result builders

```python
# Success
skill_success(message, *, prompt=None, **context) -> dict

# Failure — explicit error string
skill_error(message, error, *, prompt=None, possible_solutions=None, **context) -> dict

# Success with a warning note in context["warning"]
skill_warning(message, *, warning="", prompt=None, **context) -> dict

# Failure built from a caught exception (captures traceback in context)
skill_exception(exc, *, message=None, prompt=None, include_traceback=True,
                possible_solutions=None, **context) -> dict
```

### @skill_entry decorator

Wraps a skill function with automatic error handling. Catches `ImportError` (DCC not
available), `Exception`, and `BaseException`. When run directly, prints JSON to stdout.

```python
from dcc_mcp_core.skill import skill_entry, skill_success

@skill_entry
def set_timeline(start_frame: float = 1.0, end_frame: float = 120.0, **kwargs):
    import maya.cmds as cmds          # ImportError caught automatically
    cmds.playbackOptions(min=start_frame, max=end_frame,
                         animationStartTime=start_frame, animationEndTime=end_frame)
    return skill_success(
        f"Timeline set to {start_frame}–{end_frame}",
        prompt="Inspect the timeline slider to verify.",
        start_frame=start_frame,
        end_frame=end_frame,
    )

def main(**kwargs):
    return set_timeline(**kwargs)

if __name__ == "__main__":
    from dcc_mcp_core.skill import run_main
    run_main(main)
```

### run_main

Executes `main_fn()`, prints JSON result to stdout, exits 0/1.

```python
if __name__ == "__main__":
    from dcc_mcp_core.skill import run_main
    run_main(main)
```

### Migration from DCC-specific helpers

| DCC-specific (dcc_mcp_maya) | Generic (dcc_mcp_core.skill) |
|-----------------------------|------------------------------|
| `maya_success(msg, **ctx)` | `skill_success(msg, **ctx)` |
| `maya_error(msg, error, **ctx)` | `skill_error(msg, error, **ctx)` |
| `maya_from_exception(exc_str, ...)` | `skill_exception(exc, ...)` |

Dict structure is identical — both compatible with `ToolResult`.

---

## Result Serialization — `serialize_result` / `deserialize_result`

Rust-backed, format-agnostic serialization for `ToolResult`.
Format is controlled by `SerializeFormat` enum — JSON now, MsgPack later, no API change.

```python
from dcc_mcp_core import serialize_result, deserialize_result, SerializeFormat, success_result
```

### SerializeFormat

```python
SerializeFormat.Json     # str  — UTF-8 JSON text (default)
SerializeFormat.MsgPack  # bytes — binary MessagePack (rmp-serde)
```

### serialize_result

```python
serialize_result(result: ToolResult, format: SerializeFormat = SerializeFormat.Json) -> str | bytes
```

- Returns `str` for `Json`, `bytes` for `MsgPack`.

```python
arm = success_result("done", count=3)
json_str = serialize_result(arm)                              # str
pack_bytes = serialize_result(arm, SerializeFormat.MsgPack)  # bytes
```

### deserialize_result

```python
deserialize_result(data: str | bytes, format: SerializeFormat = SerializeFormat.Json) -> ToolResult
```

- Accepts `str` (JSON) or `bytes` (MsgPack). Format must match serialization.

```python
arm2 = deserialize_result(json_str)
assert arm2.message == "done"
arm3 = deserialize_result(pack_bytes, SerializeFormat.MsgPack)
assert arm3.context["count"] == 3
```

### How run_main serializes

`run_main()` in `dcc_mcp_core.skill` uses this Rust path when `_core` is available:
```
dict → validate_action_result() → ToolResult → serialize_result() → JSON str → stdout
```
Falls back to `json.dumps` in pure-Python DCC environments (no `_core` installed).

### Rust API (for crate authors)

```rust
use dcc_mcp_models::{ActionResultModelData, SerializeFormat};

let data = ActionResultModelData::success("done".into(), None, Default::default());
let json_bytes = data.to_bytes(SerializeFormat::Json)?;
let msgpack_bytes = data.to_bytes(SerializeFormat::MsgPack)?;
let restored = ActionResultModelData::from_bytes(&json_bytes, SerializeFormat::Json)?;
```

---

## MCP Protocol Types (`dcc_mcp_core`)

### ToolDefinition

```python
td = dcc_mcp_core.ToolDefinition(
    name="create_sphere",
    description="Creates a polygon sphere in the active Maya scene",
    input_schema='{"type": "object", "required": ["radius"], "properties": {"radius": {"type": "number", "minimum": 0.01}}}',
    output_schema=None,         # Optional JSON Schema string
    annotations=dcc_mcp_core.ToolAnnotations(
        title="Create Sphere",
        read_only_hint=False,
        destructive_hint=False,
        idempotent_hint=False,
        open_world_hint=True,   # may affect external state
    ),
)
# All fields get/set: td.name, td.description, td.input_schema, td.output_schema, td.annotations
```

### ToolAnnotations

```python
ann = dcc_mcp_core.ToolAnnotations(
    title="Create Sphere",    # Display name for UI
    read_only_hint=True,      # True = doesn't modify state; helps LLM plan safely
    destructive_hint=False,   # True = deletes/overwrites data
    idempotent_hint=True,     # True = safe to call multiple times
    open_world_hint=False,    # True = may affect external/open-ended state
    deferred_hint=None,        # True = full schema deferred until load_skill (set by server)
)
```

### ToolDeclaration

Declaration of a single MCP tool provided by a Skill, parsed from SKILL.md frontmatter.
Lightweight discovery-time object — no script execution needed.

```python
from dcc_mcp_core import ToolDeclaration

td = ToolDeclaration(
    name="create_sphere",
    description="Create a polygon sphere",
    input_schema='{"type": "object", "required": ["radius"], "properties": {"radius": {"type": "number"}}}',
    output_schema=None,       # Optional JSON Schema string
    read_only=False,
    destructive=False,
    idempotent=False,
    source_file="scripts/create_sphere.py",
)

# All fields are get/set:
td.name          # "create_sphere"
td.description   # str
td.input_schema  # JSON Schema string
td.output_schema # str|None
td.read_only     # bool  — True = doesn't modify DCC state
td.destructive   # bool  — True = deletes / overwrites data
td.idempotent    # bool  — True = safe to call multiple times
td.source_file   # str   — relative path within skill's scripts/ dir
```

**Relation to SkillMetadata**: `skill.tools` is a `List[ToolDeclaration]` populated from the
`tools:` array in SKILL.md frontmatter (v0.12+ format). Each entry corresponds to one MCP tool
that the skill exposes. The `source_file` field links the declaration back to the script that
implements it.

### ResourceDefinition / ResourceTemplateDefinition

```python
rd = dcc_mcp_core.ResourceDefinition(
    uri="file:///project/scene.mb",
    name="current-scene",
    description="The currently open Maya scene file",
    mime_type="application/octet-stream",  # default: "text/plain"
    annotations=dcc_mcp_core.ResourceAnnotations(
        audience=["user", "assistant"],
        priority=0.9,  # 0.0-1.0
    ),
)

# Template (URI with {placeholders})
rtd = dcc_mcp_core.ResourceTemplateDefinition(
    uri_template="maya://scene/{scene_name}/objects",
    name="scene-objects",
    description="Objects in a named Maya scene",
    mime_type="application/json",
)
```

### PromptDefinition / PromptArgument

```python
arg1 = dcc_mcp_core.PromptArgument("object_name", "Name of the 3D object", required=True)
arg2 = dcc_mcp_core.PromptArgument("format", "Export format", required=False)

pd = dcc_mcp_core.PromptDefinition(
    name="review_model",
    description="Review a 3D model and suggest improvements",
    arguments=[arg1, arg2],
)
```

### DCC Adapter Types

```python
# DccInfo — information about a running DCC instance
info = dcc_mcp_core.DccInfo(
    dcc_type="maya",
    version="2025",
    platform="windows",
    pid=12345,
    python_version="3.11.0",
    metadata={"scene": "/project/scene.mb"},
)
info.to_dict()  # -> dict

# DccCapabilities — what features the DCC adapter supports
caps = dcc_mcp_core.DccCapabilities(
    script_languages=[dcc_mcp_core.ScriptLanguage.PYTHON, dcc_mcp_core.ScriptLanguage.MEL],
    scene_info=True,
    snapshot=True,
    undo_redo=True,
    progress_reporting=False,
    file_operations=True,
    selection=True,
    # Cross-DCC protocol trait flags (v0.12+):
    scene_manager=True,    # implements DccSceneManager (scene/file management)
    transform=True,        # implements DccTransform (object TRS transforms)
    render_capture=False,  # implements DccRenderCapture (viewport capture + render)
    hierarchy=True,        # implements DccHierarchy (parent/child hierarchy)
)

# DccError — structured DCC error
err = dcc_mcp_core.DccError(
    code=dcc_mcp_core.DccErrorCode.SCRIPT_ERROR,
    message="Python script raised an exception",
    details="AttributeError: 'NoneType' ...",
    recoverable=True,
)

# ScriptLanguage enum values
dcc_mcp_core.ScriptLanguage.PYTHON
dcc_mcp_core.ScriptLanguage.MEL
dcc_mcp_core.ScriptLanguage.MAXSCRIPT
dcc_mcp_core.ScriptLanguage.HSCRIPT    # Houdini
dcc_mcp_core.ScriptLanguage.VEX        # Houdini VEX
dcc_mcp_core.ScriptLanguage.LUA
dcc_mcp_core.ScriptLanguage.CSHARP
dcc_mcp_core.ScriptLanguage.BLUEPRINT  # Unreal Engine

# DccErrorCode enum values
dcc_mcp_core.DccErrorCode.CONNECTION_FAILED
dcc_mcp_core.DccErrorCode.TIMEOUT
dcc_mcp_core.DccErrorCode.SCRIPT_ERROR
dcc_mcp_core.DccErrorCode.NOT_RESPONDING
dcc_mcp_core.DccErrorCode.UNSUPPORTED
dcc_mcp_core.DccErrorCode.PERMISSION_DENIED
dcc_mcp_core.DccErrorCode.INVALID_INPUT
dcc_mcp_core.DccErrorCode.SCENE_ERROR
dcc_mcp_core.DccErrorCode.INTERNAL
```

---

## Transport Layer (`dcc_mcp_core`)

### TransportAddress

Protocol-agnostic endpoint. Supports TCP, Windows Named Pipes, Unix Domain Sockets.

```python
from dcc_mcp_core import TransportAddress

# Factory methods
addr = TransportAddress.tcp("127.0.0.1", 18812)
addr = TransportAddress.named_pipe("dcc-mcp-maya")        # Windows
addr = TransportAddress.unix_socket("/tmp/dcc-mcp.sock")  # Unix
addr = TransportAddress.default_local("maya", pid=12345)  # Best for current platform
addr = TransportAddress.parse("tcp://127.0.0.1:18812")    # from URI string

# Properties
addr.scheme         # "tcp" | "pipe" | "unix"
addr.is_local       # bool
addr.is_tcp, addr.is_named_pipe, addr.is_unix_socket  # bool
addr.to_connection_string()  # "tcp://127.0.0.1:18812"
```

### TransportScheme

Transport selection strategy enum. Chooses the optimal communication channel based on platform
and availability.

```python
from dcc_mcp_core import TransportScheme, TransportAddress

# Enum values
TransportScheme.AUTO                # Automatically pick best available transport
TransportScheme.TCP_ONLY            # Force TCP (cross-machine or when IPC unavailable)
TransportScheme.PREFER_NAMED_PIPE   # Prefer Windows Named Pipes (falls back to TCP)
TransportScheme.PREFER_UNIX_SOCKET  # Prefer Unix Domain Sockets (falls back to TCP)
TransportScheme.PREFER_IPC          # Prefer any IPC (Named Pipe on Windows, Unix Socket on Linux/macOS)

# Select an address using the strategy
scheme = TransportScheme.PREFER_IPC
addr = scheme.select_address(dcc_type="maya", host="127.0.0.1", port=18812, pid=12345)
# -> TransportAddress (Named Pipe on Windows, Unix Socket on Linux/macOS, TCP fallback)
```

### IpcListener + ListenerHandle

Server-side: bind a listener and accept connections.

```python
from dcc_mcp_core import IpcListener, TransportAddress

# Bind
addr = TransportAddress.tcp("127.0.0.1", 0)   # port 0 = OS assigns
listener = IpcListener.bind(addr)
local_addr = listener.local_address()          # get assigned port
print(f"Listening on {local_addr.to_connection_string()}")

# Accept one connection (blocking)
channel = listener.accept(timeout_ms=10000)   # -> FramedChannel

# Or convert to handle for connection tracking
handle = listener.into_handle()  # consumes listener
print(handle.accept_count)       # 0 initially
print(handle.is_shutdown)        # False
handle.shutdown()                # request stop
```

### connect_ipc (client)

```python
from dcc_mcp_core import connect_ipc, TransportAddress

addr = TransportAddress.tcp("127.0.0.1", 18812)
channel = connect_ipc(addr, timeout_ms=10000)  # -> FramedChannel
```

### FramedChannel

Full-duplex message-framed channel. Handles Ping/Pong heartbeats automatically.

```python
# Ping (round-trip latency check)
rtt_ms = channel.ping(timeout_ms=5000)  # -> int

# RPC-style call (send Request, wait for matching Response)
result = channel.call(
    "execute_python",
    params=b'import maya.cmds as cmds; cmds.sphere()',
    timeout_ms=30000,
)
# result: {"id": "...", "success": bool, "payload": bytes, "error": str|None}
if result["success"]:
    output = result["payload"]
else:
    raise RuntimeError(result["error"])

# Low-level send/recv
req_id = channel.send_request("execute_mel", params=b'sphere -r 1;')
channel.send_notify("scene_changed", data=b'{"file": "scene.mb"}')
msg = channel.recv(timeout_ms=5000)   # -> dict|None (blocks)
msg = channel.try_recv()              # -> dict|None (non-blocking)

# Shutdown
channel.shutdown()       # graceful, idempotent
channel.is_running       # bool

# Message dict structure from recv():
# Request:  {"type": "request",  "id": str, "method": str, "params": bytes|None}
# Response: {"type": "response", "id": str, "success": bool, "payload": bytes|None, "error": str|None}
# Notify:   {"type": "notify",   "topic": str, "data": bytes|None}
```

### TransportManager

Full-featured manager with service discovery, session management, and connection pooling.

```python
from dcc_mcp_core import TransportManager, ServiceStatus, RoutingStrategy

mgr = TransportManager(
    registry_dir="/tmp/dcc-mcp-registry",
    max_connections_per_dcc=10,
    idle_timeout=300,           # seconds
    heartbeat_interval=5,       # seconds
    connect_timeout=10,         # seconds
    reconnect_max_retries=3,
)

# Service discovery
instance_id = mgr.register_service(
    dcc_type="maya",
    host="127.0.0.1",
    port=18812,
    version="2025",
    scene="/project/scene.mb",
    metadata={"user": "artist1"},
)
mgr.deregister_service("maya", instance_id)
instances = mgr.list_instances("maya")           # -> List[ServiceEntry]
all_services = mgr.list_all_services()            # -> List[ServiceEntry]
all_services = mgr.list_all_instances()           # alias for list_all_services()
entry = mgr.get_service("maya", instance_id)      # -> ServiceEntry|None

# Heartbeat + status
mgr.heartbeat("maya", instance_id)  # -> bool (False if not found)
mgr.update_service_status("maya", instance_id, ServiceStatus.BUSY)

# One-call server registration (v0.12+) — bind listener + register in one step
instance_id, listener = mgr.bind_and_register(
    "maya",
    version="2025",
    metadata={"artist": "user1"},
)
# Transport auto-selected: Named Pipe on Windows, Unix Socket on Linux/macOS, TCP fallback
local_addr = listener.local_address()
channel = listener.accept()   # wait for first client connection

# Smart service discovery (v0.12+)
best = mgr.find_best_service("maya")      # best available instance (raises if none)
ranked = mgr.rank_services("maya")        # all live instances sorted by preference
# Preference order: Local IPC AVAILABLE → Local IPC BUSY → Local TCP AVAIL → Remote TCP ...
for entry in ranked:
    print(entry.instance_id, entry.status, entry.effective_address())

# Session management (automatic connection routing)
session_id = mgr.get_or_create_session("maya", instance_id)
session_id = mgr.get_or_create_session_routed(
    "maya",
    strategy=RoutingStrategy.LEAST_BUSY,
)
session = mgr.get_session(session_id)      # -> dict|None
mgr.record_success(session_id, latency_ms=42)
mgr.record_error(session_id, latency_ms=500, error="timeout")
mgr.close_session(session_id)              # -> bool

# Connection pool
conn_id = mgr.acquire_connection("maya", instance_id)
mgr.release_connection("maya", instance_id)
mgr.pool_size()                 # -> int
mgr.pool_count_for_dcc("maya")  # -> int

# Lifecycle
stats = mgr.cleanup()   # -> (expired_sessions, idle_conns, errors) tuple
mgr.shutdown()
mgr.is_shutdown()       # -> bool
len(mgr)                # total managed connections

# ServiceStatus values
ServiceStatus.AVAILABLE
ServiceStatus.BUSY
ServiceStatus.UNREACHABLE
ServiceStatus.SHUTTING_DOWN

# RoutingStrategy values
RoutingStrategy.FIRST_AVAILABLE
RoutingStrategy.ROUND_ROBIN
RoutingStrategy.LEAST_BUSY
RoutingStrategy.SPECIFIC
RoutingStrategy.SCENE_MATCH
RoutingStrategy.RANDOM
```

### ServiceEntry

```python
entry.dcc_type           # str
entry.instance_id        # str
entry.host, entry.port   # str, int
entry.version            # str|None
entry.scene              # str|None (current open scene path)
entry.metadata           # dict[str,str]
entry.status             # ServiceStatus
entry.transport_address  # TransportAddress|None
entry.last_heartbeat_ms  # int (Unix ms; updated by TransportManager.heartbeat())
entry.is_ipc             # bool
entry.effective_address()  # -> TransportAddress (transport_address or TCP fallback)
entry.to_dict()

# Check staleness
import time
idle_sec = (time.time() * 1000 - entry.last_heartbeat_ms) / 1000
if idle_sec > 300:
    mgr.deregister_service(entry.dcc_type, entry.instance_id)
```

---

## Cross-DCC Protocol Data Models (`dcc_mcp_core`)

Coordinate-system–normalized data models for DCC-agnostic communication.
All adapters (Maya, Blender, Houdini, 3dsMax, Unreal, Unity, Photoshop, Figma, …)
convert from their native representation to these shared types.

### ObjectTransform

3D object TRS in right-hand Y-up world space.

```python
from dcc_mcp_core import ObjectTransform

t = ObjectTransform(
    translate=[0.0, 10.0, 0.0],  # [x, y, z] in centimeters
    rotate=[0.0, 45.0, 0.0],     # Euler XYZ in degrees
    scale=[1.0, 1.0, 1.0],       # [sx, sy, sz]
)
t.translate   # [float, float, float]
t.rotate      # [float, float, float]
t.scale       # [float, float, float]
ObjectTransform.identity()  # -> ObjectTransform (all zeros/ones)
t.to_dict()   # -> {"translate": [...], "rotate": [...], "scale": [...]}
```

### BoundingBox

Axis-aligned bounding box in world space (centimeters).

```python
from dcc_mcp_core import BoundingBox

bb = BoundingBox(min=[-1.0, 0.0, -1.0], max=[1.0, 2.0, 1.0])
bb.min, bb.max  # list[float], list[float]
bb.center()     # -> [0.0, 1.0, 0.0]
bb.size()       # -> [2.0, 2.0, 2.0]
bb.to_dict()    # -> {"min": [...], "max": [...]}
```

### SceneObject

Lightweight descriptor of any scene entity (mesh, light, camera, transform, layer, actor, …).

```python
from dcc_mcp_core import SceneObject

obj = SceneObject(
    name="pCube1",
    long_name="|group1|pCube1",  # full DAG / hierarchy path
    object_type="mesh",          # "mesh", "light", "camera", "transform", etc.
    parent="group1",             # parent name (optional)
    visible=True,
    metadata={"material": "lambert1"},
)
obj.name, obj.long_name, obj.object_type
obj.parent   # str|None
obj.visible  # bool
obj.to_dict()
```

### SceneNode

Scene hierarchy node with recursive children (DAG node / actor / layer group).

```python
from dcc_mcp_core import SceneNode, SceneObject

leaf = SceneNode(object=SceneObject(name="pSphere1", object_type="mesh"))
root = SceneNode(
    object=SceneObject(name="group1", object_type="transform"),
    children=[leaf],
)
root.object    # SceneObject
root.children  # list[SceneNode]
root.to_dict()
```

### FrameRange

Animation frame range and timing.

```python
from dcc_mcp_core import FrameRange

fr = FrameRange(start=1.0, end=240.0, fps=24.0, current=1.0)
fr.start, fr.end, fr.fps, fr.current  # float
duration_seconds = (fr.end - fr.start + 1) / fr.fps
fr.to_dict()
```

### RenderOutput

Metadata for a completed render operation.

```python
from dcc_mcp_core import RenderOutput

out = RenderOutput(
    file_path="/renders/frame001.png",
    width=1920,
    height=1080,
    format="png",
    render_time_ms=5000,
)
out.file_path, out.width, out.height, out.format, out.render_time_ms
out.to_dict()
```

---

## Process Management (`dcc_mcp_core`)

### PyDccLauncher

Async DCC process launcher (spawn / terminate / kill).

```python
from dcc_mcp_core import PyDccLauncher

launcher = PyDccLauncher()

# Launch a DCC process
info = launcher.launch(
    name="maya-2025",                        # logical name for tracking
    executable="/usr/autodesk/maya/bin/maya",
    args=["-batch", "-file", "scene.mb"],
    launch_timeout_ms=30000,
)
# info: {"pid": 12345, "name": "maya-2025", "status": "running"}

# Manage
launcher.terminate("maya-2025", timeout_ms=5000)  # graceful SIGTERM
launcher.kill("maya-2025")                         # force SIGKILL
pid = launcher.pid_of("maya-2025")                 # -> int|None
count = launcher.running_count()                   # int
restarts = launcher.restart_count("maya-2025")     # int
```

### PyProcessMonitor

Cross-platform DCC process monitor (CPU/memory sampling via sysinfo).

```python
import os
from dcc_mcp_core import PyProcessMonitor

mon = PyProcessMonitor()
mon.track(os.getpid(), "self")  # register a PID to monitor
mon.refresh()                   # refresh OS data (must call before query)
info = mon.query(os.getpid())
# info: {"pid": N, "name": "self", "status": "running",
#         "cpu_usage_percent": 1.2, "memory_bytes": 123456, "restart_count": 0}

mon.list_all()    # -> List[dict] for all tracked PIDs
mon.is_alive(N)   # -> bool (PID in OS process table)
mon.tracked_count()  # -> int
mon.untrack(N)    # stop monitoring
```

### PyProcessWatcher + PyCrashRecoveryPolicy

Background process watcher with event polling (async under the hood).

```python
import os, time
from dcc_mcp_core import PyProcessWatcher, PyCrashRecoveryPolicy

# Configure crash recovery
policy = PyCrashRecoveryPolicy(max_restarts=3)
policy.use_exponential_backoff(initial_ms=1000, max_delay_ms=30000)
# Or: policy.use_fixed_backoff(delay_ms=2000)
policy.should_restart("crashed")      # -> bool
policy.should_restart("unresponsive") # -> bool
delay = policy.next_delay_ms("maya", attempt=0)  # -> int ms

# Watcher
watcher = PyProcessWatcher(poll_interval_ms=500)
watcher.track(os.getpid(), "self")
watcher.start()

time.sleep(1.0)
events = watcher.poll_events()  # drain all pending events -> List[dict]
# Event dict keys: "type", "pid", "name"
# "heartbeat":      + "new_status", "cpu_usage_percent", "memory_bytes"
# "status_changed": + "old_status", "new_status"
# "exited":         (just type/pid/name)

watcher.stop()
watcher.is_running()       # -> bool
watcher.tracked_count()    # -> int
watcher.untrack(N)
```

### ScriptResult

Result of executing a DCC script.

```python
from dcc_mcp_core import ScriptResult

r = ScriptResult(
    success=True,
    execution_time_ms=42,
    output="sphere1",
    error=None,
    context={"dcc": "maya"},
)
r.to_dict()
```

---

## Sandbox (`dcc_mcp_core`)

### SandboxPolicy

API whitelist, path allowlist, execution constraints.

```python
from dcc_mcp_core import SandboxPolicy

policy = SandboxPolicy()
policy.allow_actions(["get_scene_info", "list_objects", "create_sphere"])
policy.deny_actions(["delete_scene", "shutdown"])  # override allowlist
policy.allow_paths(["/project/assets", "/project/cache"])
policy.set_timeout_ms(5000)
policy.set_max_actions(100)
policy.set_read_only(False)

policy.is_read_only  # bool
```

### SandboxContext

Bundles policy + audit log + action counter.

```python
from dcc_mcp_core import SandboxPolicy, SandboxContext

policy = SandboxPolicy()
policy.allow_actions(["echo"])
ctx = SandboxContext(policy)
ctx.set_actor("ai-agent-v1")   # identifies caller in audit log

result_json = ctx.execute_json("echo", '{"x": 1}')  # -> str (JSON result)
# Raises RuntimeError if denied/timeout/error

ctx.action_count   # int (successful executions)
ctx.audit_log      # AuditLog instance

ctx.is_allowed("echo")              # -> bool
ctx.is_path_allowed("/project/assets")  # -> bool
```

### InputValidator

Schema-based input validator with injection-guard rules.

```python
from dcc_mcp_core import InputValidator

v = InputValidator()
v.require_string("name", max_length=50, min_length=1)
v.require_number("count", min_value=0, max_value=1000)
v.forbid_substrings("script", ["__import__", "exec(", "eval(", "os.system"])

ok, error = v.validate('{"name": "sphere", "count": 5}')
# ok=True, error=None

ok, error = v.validate('{"script": "__import__(os)"}')
# ok=False, error="..."
```

### AuditLog + AuditEntry

```python
log = ctx.audit_log

len(log)            # total entries
entries = log.entries()     # -> List[AuditEntry]
successes = log.successes() # only successful entries
denials = log.denials()     # only denied entries
action_entries = log.entries_for_action("create_sphere")
log_json = log.to_json()    # -> str (JSON array)

# AuditEntry fields (all read-only properties)
entry.timestamp_ms    # Unix ms
entry.actor           # str|None (set via ctx.set_actor())
entry.action          # str (action name)
entry.params_json     # str (parameters as JSON)
entry.duration_ms     # int
entry.outcome         # "success"|"denied"|"error"|"timeout"
entry.outcome_detail  # str|None (error message or denial reason)
```

---

## Telemetry (`dcc_mcp_core`)

### ToolRecorder + RecordingGuard

Per-action timing and success/failure counters.

```python
from dcc_mcp_core import ToolRecorder

recorder = ToolRecorder("maya-mcp-server")  # scope name

# Method 1: manual guard
guard = recorder.start("create_sphere", "maya")
try:
    result = do_create_sphere()
    guard.finish(success=True)
except Exception:
    guard.finish(success=False)
    raise

# Method 2: context manager (exception → success=False)
with recorder.start("batch_rename", "maya") as guard:
    do_batch_rename()
# guard.finish() called automatically with success=(no exception)

# Query metrics
metrics = recorder.metrics("create_sphere")   # -> ToolMetrics|None
all_m = recorder.all_metrics()               # -> List[ToolMetrics]
recorder.reset()                             # clear in-memory stats
```

### ToolMetrics

Read-only snapshot of per-action performance.

```python
m = recorder.metrics("create_sphere")
m.action_name         # str
m.invocation_count    # int (total calls)
m.success_count       # int
m.failure_count       # int
m.avg_duration_ms     # float
m.p95_duration_ms     # float (95th percentile)
m.p99_duration_ms     # float (99th percentile)
m.success_rate()      # -> float in [0.0, 1.0]
```

### TelemetryConfig

Global telemetry provider (OpenTelemetry-backed).

```python
from dcc_mcp_core import TelemetryConfig, is_telemetry_initialized, shutdown_telemetry

cfg = (TelemetryConfig("maya-mcp-server")
       .with_stdout_exporter()          # print spans to stdout
       .with_attribute("dcc.type", "maya")
       .with_attribute("dcc.version", "2025")
       .with_service_version("1.0.0")
       .set_enable_metrics(True)
       .set_enable_tracing(True))
cfg.init()  # install as global provider (raises if already initialized)

is_telemetry_initialized()  # -> bool
shutdown_telemetry()        # flush and shut down

# For tests (no output):
cfg = TelemetryConfig("test").with_noop_exporter()
# For structured logs:
cfg = TelemetryConfig("prod").with_json_logs()
```

---

## Shared Memory (`dcc_mcp_core`)

Zero-copy data exchange for large DCC scene data (mesh geometry, animation caches, screenshots).

### PySharedBuffer

Named, fixed-capacity shared memory buffer backed by a memory-mapped file.

```python
from dcc_mcp_core import PySharedBuffer

# Create (producer side)
buf = PySharedBuffer.create(capacity=1024 * 1024)  # 1 MiB
n_written = buf.write(b"vertex data bytes")
data = buf.read()

buf.id           # str (UUID)
buf.path()       # str (file path of backing mmap)
buf.data_len()   # int (bytes currently stored)
buf.capacity()   # int (max bytes)
buf.clear()      # reset to 0 bytes
desc = buf.descriptor_json()  # JSON for cross-process handoff

# Open on consumer side (using producer's descriptor info)
buf2 = PySharedBuffer.open(path=buf.path(), id=buf.id)
assert buf2.read() == b"vertex data bytes"
```

### PyBufferPool

Fixed-capacity pool of reusable shared memory buffers. Amortises mmap allocation cost.

```python
from dcc_mcp_core import PyBufferPool

pool = PyBufferPool(capacity=4, buffer_size=1024 * 1024)  # 4 × 1MiB
buf = pool.acquire()   # -> PySharedBuffer (raises if all slots in use)
buf.write(b"scene snapshot")
# Buffer returned to pool automatically when garbage-collected

pool.available()      # -> int (free slots)
pool.capacity()       # -> int (total)
pool.buffer_size()    # -> int (per-buffer bytes)
```

### PySharedSceneBuffer (High-level)

High-level wrapper. Automatically selects inline (< 256 MiB) vs chunked (≥ 256 MiB) storage.
Supports optional LZ4 compression.

```python
from dcc_mcp_core import PySharedSceneBuffer, PySceneDataKind

# Write (producer DCC side)
ssb = PySharedSceneBuffer.write(
    data=vertex_bytes,
    kind=PySceneDataKind.Geometry,
    source_dcc="Maya",
    use_compression=True,   # LZ4
)
desc_json = ssb.descriptor_json()  # send this to consumer via IPC channel

# Read (consumer agent side) — reconstruct from descriptor is not exposed;
# pass ssb itself if in-process, or use FramedChannel to send desc_json
recovered = ssb.read()  # -> bytes (decompresses automatically)

ssb.id            # str
ssb.total_bytes   # int (original uncompressed size)
ssb.is_inline     # bool
ssb.is_chunked    # bool

# PySceneDataKind values
PySceneDataKind.Geometry
PySceneDataKind.AnimationCache
PySceneDataKind.Screenshot
PySceneDataKind.Arbitrary
```

---

## Screen Capture (`dcc_mcp_core`)

### Capturer + CaptureFrame

High-level DCC viewport / desktop capture. Platform-specific backends:
- Windows (full-screen): DXGI Desktop Duplication (GPU framebuffer, < 16 ms/frame)
- Windows (single window): HWND `PrintWindow` + `BitBlt` fallback
- Linux: X11 XShmGetImage (full-screen) — PipeWire reserved for Wayland
- macOS: ScreenCaptureKit (reserved)
- Fallback: Mock synthetic checkerboard (CI / headless)

```python
from dcc_mcp_core import Capturer, CaptureBackendKind

# Auto-select best full-screen / display backend
capturer = Capturer.new_auto()

# Auto-select best single-window backend (HWND on Windows, Mock elsewhere)
window_cap = Capturer.new_window_auto()

# Mock backend (headless CI, no GPU needed)
capturer = Capturer.new_mock(width=1920, height=1080)

# Capture a frame
frame = capturer.capture(
    format="png",           # "png" | "jpeg" | "raw_bgra"
    jpeg_quality=85,        # 0-100, only for jpeg
    scale=0.5,              # 0.0-1.0 scale factor
    timeout_ms=5000,
    process_id=12345,       # capture specific window by PID
    window_title="Maya",    # capture window by title substring
)

# Or capture a single window explicitly
frame = window_cap.capture_window(
    window_title="Autodesk Maya 2024",
    include_decorations=True,
    format="png",
    timeout_ms=5000,
)
# window_handle=<HWND int> and process_id=<pid> are also accepted

# CaptureFrame fields
frame.data           # bytes (PNG/JPEG/raw)
frame.width          # int
frame.height         # int
frame.format         # "png" | "jpeg" | "raw_bgra"
frame.mime_type      # "image/png" | "image/jpeg"
frame.timestamp_ms   # int (Unix ms)
frame.dpi_scale      # float (1.0=normal, 2.0=HiDPI)
frame.window_rect    # (x, y, w, h) tuple or None for full-screen captures
frame.window_title   # str or None
frame.byte_len()     # int

# Save to disk
with open("screenshot.png", "wb") as f:
    f.write(frame.data)

# Stats / backend introspection
capturer.backend_name()                                 # "DXGI Desktop Duplication" | "HWND PrintWindow" | "X11" | "Mock"
capturer.backend_kind() == CaptureBackendKind.HwndPrintWindow
count, total_bytes, errors = capturer.stats()
```

### CaptureTarget + WindowFinder

Resolve windows / displays to a capture-ready descriptor without throwing when
the target is missing.

```python
from dcc_mcp_core import CaptureTarget, WindowFinder

# Static factories — opaque descriptor, no raise on invalid
CaptureTarget.primary_display()
CaptureTarget.monitor_index(0)
CaptureTarget.process_id(12345)
CaptureTarget.window_title("Maya 2024")
CaptureTarget.window_handle(0x00A1B2)

finder = WindowFinder()
info = finder.find(CaptureTarget.process_id(12345))
if info is not None:
    info.handle   # int (HWND / X11 window id)
    info.pid      # int
    info.title    # str
    info.rect     # (x, y, w, h)

# Enumerate every visible top-level window on Windows
for info in finder.enumerate():
    print(info.handle, info.title)
```

### Instance-Bound Diagnostics

When multiple DCC instances run side-by-side, each `DccServerBase` subclass
should be bound to **its own** DCC process so `diagnostics__*` MCP tools and
the bundled `dcc-diagnostics` skill target the right window / PID.

```python
from dcc_mcp_core import DccServerBase

class MayaServer(DccServerBase):
    def __init__(self, pid: int, window_title: str):
        super().__init__(
            dcc_name="maya",
            builtin_skills_dir=None,
            dcc_pid=pid,                     # owner DCC PID
            dcc_window_title=window_title,   # title fallback for window lookups
            # dcc_window_handle=0x00A1B2,    # or pass an HWND directly
            # resolver=lambda: current_maya_pid(),  # late-bound PID
        )

server = MayaServer(pid=12345, window_title="Autodesk Maya 2024")
handle = server.start()
# Registers the four diagnostics tools bound to this instance:
#   diagnostics__screenshot
#   diagnostics__audit_log
#   diagnostics__tool_metrics   (renamed from diagnostics__action_metrics in 0.14.0)
#   diagnostics__process_status
```

For low-level servers that build on `McpHttpServer` directly, call
`register_diagnostic_mcp_tools(server, *, dcc_name=..., dcc_pid=..., ...)` or
`register_diagnostic_handlers(...)` (IPC path) **before** `server.start()` —
MCP tool registration must complete before the server is running.

The bundled `dcc-diagnostics` skill's `screenshot.py` handler tries
`DCC_MCP_OWNER_IPC` → `channel.call("take_screenshot")` first and falls back
to `Capturer.new_auto()` when no owner IPC is configured.

---

## USD Scene Description (`dcc_mcp_core`)

Lightweight USD-like scene exchange format for DCC interoperability.

### SdfPath

```python
from dcc_mcp_core import SdfPath

path = SdfPath("/World")
child = path.child("Cube")  # -> SdfPath("/World/Cube")
parent = child.parent()     # -> SdfPath("/World")
child.name                  # "Cube"
child.is_absolute           # True
hash(child)                 # hashable for use as dict key
```

### VtValue

Variant value container for USD attributes.

```python
from dcc_mcp_core import VtValue

# Create from Python types
v_bool   = VtValue.from_bool(True)
v_int    = VtValue.from_int(42)
v_float  = VtValue.from_float(1.0)
v_str    = VtValue.from_string("Cube")
v_token  = VtValue.from_token("Mesh")    # USD token (interned string)
v_asset  = VtValue.from_asset("/path/to/file.usd")
v_vec3   = VtValue.from_vec3f(1.0, 2.0, 3.0)

v_float.type_name          # "float"
v_vec3.type_name           # "float3"
v_float.to_python()        # 1.0
v_vec3.to_python()         # (1.0, 2.0, 3.0) tuple
```

### UsdPrim

```python
from dcc_mcp_core import UsdStage, UsdPrim, VtValue

stage = UsdStage("my_scene")
prim = stage.define_prim("/World/Cube", "Mesh")  # -> UsdPrim

prim.path         # SdfPath("/World/Cube")
prim.type_name    # "Mesh"
prim.active       # bool
prim.name         # "Cube"

prim.set_attribute("extent", VtValue.from_vec3f(1, 1, 1))
val = prim.get_attribute("extent")   # -> VtValue|None
names = prim.attribute_names()       # -> List[str]
summary = prim.attributes_summary()  # -> dict[str, str]  (attr_name → type_name)
prim.has_api("UsdGeomModelAPI")      # -> bool
```

### UsdStage

Primary unit of cross-DCC scene exchange.

```python
from dcc_mcp_core import UsdStage, VtValue

stage = UsdStage("my_scene")

# Stage properties (get/set)
stage.name              # str
stage.id                # str (UUID)
stage.up_axis           # "Y" (default) | "Z"
stage.up_axis = "Z"
stage.meters_per_unit   # float (default 0.01 = centimeters)
stage.fps               # float|None
stage.start_time_code   # float|None
stage.end_time_code     # float|None
stage.default_prim      # str|None

# Prim operations
stage.define_prim("/World/Cube", "Mesh")  # -> UsdPrim
stage.get_prim("/World/Cube")             # -> UsdPrim|None
stage.has_prim("/World/Cube")             # -> bool
stage.remove_prim("/World/Cube")          # -> bool
stage.traverse()                          # -> List[UsdPrim] (all prims)
stage.prims_of_type("Mesh")              # -> List[UsdPrim]

# Attribute shortcuts
stage.set_attribute("/World/Cube", "extent", VtValue.from_vec3f(1, 1, 1))
stage.get_attribute("/World/Cube", "extent")  # -> VtValue|None

# Stats
stage.metrics()  # -> {"prim_count": N, "attribute_count": M, ...}

# Serialization
json_str = stage.to_json()
stage2 = UsdStage.from_json(json_str)
usda = stage.export_usda()   # USD ASCII format string
```

### USD Bridge Functions

Convert between DCC `SceneInfo` and `UsdStage`.

```python
from dcc_mcp_core import scene_info_json_to_stage, stage_to_scene_info_json, units_to_mpu, mpu_to_units

# Convert SceneInfo JSON (from DCC adapter) to UsdStage
stage = scene_info_json_to_stage(scene_info_json_str, dcc_type="maya")

# Convert UsdStage back to SceneInfo JSON (best-effort)
scene_info_json_str = stage_to_scene_info_json(stage)

# Unit conversion
mpu = units_to_mpu("cm")   # -> 0.01 (meters per unit for centimeters)
mpu = units_to_mpu("m")    # -> 1.0
units = mpu_to_units(0.01) # -> "cm"
units = mpu_to_units(1.0)  # -> "m"
```

---

## Bridge System (`dcc_mcp_core`)

DCC inter-protocol bridging — register named bridges between different transport protocols.

```python
from dcc_mcp_core import BridgeRegistry, BridgeContext, register_bridge, get_bridge_context

# Register a named bridge
register_bridge("rpyc", BridgeContext(name="rpyc", description="RPyC ↔ MCP bridge"))

# Retrieve bridge context by name
ctx = get_bridge_context("rpyc")  # -> BridgeContext or None
```

---

## Scene Data Model (`dcc_mcp_core`)

Structured scene data types for representing DCC scene hierarchy and geometry.

```python
from dcc_mcp_core import BoundingBox, FrameRange, ObjectTransform, SceneNode, SceneObject, RenderOutput

# Bounding box (axis-aligned)
bbox = BoundingBox(min_x=0, min_y=0, min_z=0, max_x=1, max_y=1, max_z=1)

# Animation frame range
frames = FrameRange(start=1, end=24, step=1)

# Object transform (translate, rotate, scale decomposition)
xform = ObjectTransform(translate=[0, 5, 0], rotate=[0, 0, 0, 1], scale=[1, 1, 1])

# Scene nodes (hierarchical)
node = SceneNode(path="/world/geo/sphere1", transform=xform)

# Scene objects (full representation)
obj = SceneObject(name="sphere1", node_type="mesh", transform=xform)

# Render output metadata
render = RenderOutput(path="/tmp/render.exr", format="exr", width=1920, height=1080)
```

> **Note**: `BoundingBox` is an optional import — may be `None` if not available in the compiled extension.

---

## Serialization (`dcc_mcp_core`)

Transport-safe ToolResult serialization for IPC, storage, or network transfer.

```python
from dcc_mcp_core import SerializeFormat, serialize_result, deserialize_result, success_result

result = success_result("Created sphere", prompt="Add materials next", count=1)

# Serialize to bytes
data = serialize_result(result, fmt=SerializeFormat.JSON)

# Deserialize back
restored = deserialize_result(data, fmt=SerializeFormat.JSON)
assert restored.success
assert restored.message == "Created sphere"
```

---

## Type Wrappers (`dcc_mcp_core`)

For safe RPyC transport (prevents OB proxy wrapping of basic Python types).

```python
from dcc_mcp_core import wrap_value, unwrap_value, unwrap_parameters
from dcc_mcp_core import BooleanWrapper, IntWrapper, FloatWrapper, StringWrapper

# Wrap — dispatches to correct type
w = wrap_value(True)      # -> BooleanWrapper(True)
w = wrap_value(42)        # -> IntWrapper(42)
w = wrap_value(3.14)      # -> FloatWrapper(3.14)
w = wrap_value("hello")   # -> StringWrapper("hello")
w = wrap_value([1, 2])    # -> [1, 2] (passthrough for unsupported types)

# Unwrap
v = unwrap_value(BooleanWrapper(True))  # -> True
v = unwrap_value(42)                    # -> 42 (passthrough)

# Bulk unwrap
result = unwrap_parameters({"enabled": BooleanWrapper(True), "count": IntWrapper(5)})
# -> {"enabled": True, "count": 5}

# Direct construction
BooleanWrapper(True).value    # True
IntWrapper(42).value          # 42; also supports __int__, __index__
FloatWrapper(3.14).value      # 3.14; also supports __float__
StringWrapper("hi").value     # "hi"; also supports __str__

# All wrappers: __repr__, __eq__, __hash__ (StringWrapper and BooleanWrapper and IntWrapper)
```

---

## Utilities (`dcc_mcp_core`)

### Filesystem Functions

```python
from dcc_mcp_core import (
    get_config_dir, get_data_dir, get_log_dir, get_platform_dir,
    get_tools_dir, get_skills_dir, get_skill_paths_from_env,
    get_app_skill_paths_from_env,
)

get_config_dir()              # Platform-specific config dir (e.g. %APPDATA%/dcc-mcp on Windows)
get_data_dir()                # Platform-specific data dir
get_log_dir()                 # Data dir + /log subdirectory
get_platform_dir("config")    # Generic: accepts "config"|"data"|"cache"|"log"|"state"|"documents"
get_tools_dir("maya")       # -> .../data/actions/maya/
get_skills_dir("maya")        # -> .../data/skills/maya/
get_skills_dir()              # -> .../data/skills/
get_skill_paths_from_env()    # -> List[str] from DCC_MCP_SKILL_PATHS env var
get_app_skill_paths_from_env("maya")   # -> List[str] from DCC_MCP_MAYA_SKILL_PATHS env var
get_app_skill_paths_from_env("blender") # -> List[str] from DCC_MCP_BLENDER_SKILL_PATHS env var
```

---

## Constants (`dcc_mcp_core`)

| Constant | Value | Purpose |
|----------|-------|---------|
| `APP_NAME` | `"dcc-mcp"` | Application name |
| `APP_AUTHOR` | `"dcc-mcp"` | Application author |
| `DEFAULT_DCC` | `"python"` | Default DCC type when none specified |
| `DEFAULT_LOG_LEVEL` | `"DEBUG"` | Default log level |
| `DEFAULT_MIME_TYPE` | `"text/plain"` | Default MIME type for resources |
| `DEFAULT_VERSION` | `"1.0.0"` | Default action version |
| `SKILL_METADATA_FILE` | `"SKILL.md"` | Skill package manifest filename |
| `SKILL_SCRIPTS_DIR` | `"scripts"` | Skill scripts subdirectory name |
| `SKILL_METADATA_DIR` | `"metadata"` | Skill metadata subdirectory name |
| `ENV_SKILL_PATHS` | `"DCC_MCP_SKILL_PATHS"` | Environment variable for skill search paths |
| `ENV_LOG_LEVEL` | `"MCP_LOG_LEVEL"` | Environment variable for log level override |

---

## Environment Variables

| Variable | Description | Example |
|----------|-------------|---------|
| `DCC_MCP_SKILL_PATHS` | Skill search paths (`:` on Unix, `;` on Windows) | `/skills1:/skills2` |
| `MCP_LOG_LEVEL` | Log level override | `INFO` / `DEBUG` / `WARN` |

---

## Typical Integration Patterns

### Pattern 1: Register skill-based actions with an MCP server

```python
import os
from dcc_mcp_core import scan_and_load, ToolRegistry, ToolDefinition, ToolAnnotations

os.environ["DCC_MCP_SKILL_PATHS"] = "/opt/skills"
skills, _ = scan_and_load(dcc_name="maya")
reg = ToolRegistry()
tools = []
for skill in skills:
    for script in skill.scripts:
        from pathlib import Path
        stem = Path(script).stem
        action_name = f"{skill.name.replace('-', '_')}__{stem}"
        reg.register(name=action_name, description=skill.description, dcc=skill.dcc)
        tools.append(ToolDefinition(
            name=action_name,
            description=skill.description,
            input_schema='{"type": "object"}',
        ))
```

### Pattern 2: Full RPC call to a running DCC

```python
import json
from dcc_mcp_core import TransportAddress, connect_ipc, success_result, error_result

addr = TransportAddress.default_local("maya", pid=12345)
channel = connect_ipc(addr)
try:
    # Primary RPC: .call() atomically sends Request + waits for correlated Response (v0.12.7+)
    result = channel.call("execute_python", b'import maya.cmds; print(maya.cmds.ls())', timeout_ms=10000)
    if result["success"]:
        payload = result.get("payload", b"")
        output = payload.decode() if isinstance(payload, bytes) else str(payload)
        r = success_result(output, prompt="Objects listed. You can now select one to modify.")
    else:
        r = error_result("DCC script failed", result.get("error", "Unknown error"))
finally:
    channel.shutdown()
```

### Pattern 3: Sandbox AI-generated code execution

```python
from dcc_mcp_core import SandboxPolicy, SandboxContext, InputValidator

policy = SandboxPolicy()
policy.allow_actions(["create_sphere", "list_objects", "get_scene_info"])
policy.allow_paths(["/project/assets"])
policy.set_timeout_ms(5000)
policy.set_max_actions(50)

ctx = SandboxContext(policy)
ctx.set_actor("claude-agent")

result_json = ctx.execute_json("create_sphere", '{"radius": 1.0, "name": "mySphere"}')
print(f"Executed {ctx.action_count} actions")
for entry in ctx.audit_log.entries():
    print(f"{entry.action}: {entry.outcome} ({entry.duration_ms}ms)")
```

### Pattern 4: Zero-copy scene data transfer

```python
from dcc_mcp_core import PySharedSceneBuffer, PySceneDataKind, TransportAddress, connect_ipc

# In DCC (producer)
mesh_bytes = get_mesh_data_as_bytes()
ssb = PySharedSceneBuffer.write(mesh_bytes, kind=PySceneDataKind.Geometry, use_compression=True)
desc = ssb.descriptor_json()
# Send desc via IPC channel to agent
channel = connect_ipc(TransportAddress.default_local("maya", os.getpid()))
channel.send_notify("mesh_ready", data=desc.encode())
```

---

## MCP HTTP Server (`dcc_mcp_core`)

The `dcc-mcp-http` crate provides a **MCP Streamable HTTP server** compliant with the **2025-03-26 MCP
specification**. It uses axum + Tokio and runs in a background thread, making it safe to call from any
DCC main thread.

> **MCP Spec Roadmap**: The MCP specification has evolved significantly since the 2025-03-26 version this library implements:
>
> | Version | Key Changes | Status |
> |---------|-------------|--------|
> | 2025-03-26 | Streamable HTTP, Tool Annotations, OAuth 2.1 | **Implemented** |
> | 2025-06-18 | Structured Tool Output, Elicitation (server asks user for info), Resource Links in tool results, JSON-RPC batching **removed**, `MCP-Protocol-Version` header mandatory | Planned |
> | 2025-11-25 | Icon metadata, Tasks (experimental), Sampling with tool calls, JSON Schema 2020-12, enhanced OAuth | Planned |
>
> **2026 MCP Roadmap** (announced March 2026) — four priority areas:
> 1. **Transport scalability**: `.well-known` server capability discovery, stateless session model for horizontal scaling
> 2. **Agent communication**: Tasks primitive lifecycle (experimental in 2025-11-25), retry/expiration semantics pending
> 3. **Governance**: contributor ladder, delegated workgroup model for faster SEP review
> 4. **Enterprise readiness**: audit trails, SSO integration, gateway behavior (mostly as extensions, not core spec changes)
>
> No new official transport types will be added in the 2026 cycle — only evolution of Streamable HTTP.
>
> When these features land in `dcc-mcp-core`, they will be exposed via `McpHttpServer` —
> do NOT implement these manually.

### McpHttpConfig

```python
from dcc_mcp_core import McpHttpConfig

config = McpHttpConfig(
    port=8765,              # use 0 for OS-assigned ephemeral port
    server_name="maya-mcp", # reported in MCP initialize response
    server_version="1.0.0",
    enable_cors=False,      # set True for browser-based MCP clients
    request_timeout_ms=30000,
)
print(config.port)          # 8765
print(config.server_name)   # "maya-mcp"
print(config.server_version) # "1.0.0"
```

```python

# Opt-in: lazy-actions fast-path — tools/list surfaces only 3 meta-tools
# (list_actions, describe_action, call_action) instead of all tools
config.lazy_actions = True  # default: False

# Gateway participation
config.gateway_port = 9765  # 0 = disabled (default)
config.dcc_type = "maya"
config.dcc_version = "2025"
config.scene = "/proj/shot01.ma"  # optional: helps routing by scene

# Session management
config.session_ttl_secs = 3600  # idle session eviction (0 = disable)
```

### McpHttpServer

```python
from dcc_mcp_core import ToolRegistry, McpHttpServer, McpHttpConfig, McpServerHandle

registry = ToolRegistry()
registry.register(
    "get_scene_info",
    description="Get info about the current scene",
    category="scene", tags=["query"], dcc="maya", version="1.0.0",
)
registry.register(
    "create_sphere",
    description="Create a polygon sphere",
    category="geometry", tags=["create"], dcc="maya", version="1.0.0",
    input_schema='{"type":"object","required":["radius"],"properties":{"radius":{"type":"number"}}}',
)

# Start server — returns immediately, server runs in background
server = McpHttpServer(registry, McpHttpConfig(port=8765, server_name="maya-mcp"))
handle = server.start()   # -> McpServerHandle (alias for McpServerHandle)
```

### McpServerHandle

```python
# McpServerHandle is the alias for McpServerHandle in __init__.py
from dcc_mcp_core import McpServerHandle

print(handle.mcp_url())      # "http://127.0.0.1:8765/mcp"  (MCP endpoint URL)
print(handle.port)           # 8765 (actual port; useful when McpHttpConfig(port=0))
print(handle.bind_addr)      # "127.0.0.1:8765"

handle.shutdown()            # blocks until server stops
handle.signal_shutdown()     # non-blocking shutdown signal
```

### create_skill_server (Skills-First, recommended)

```python
import os
os.environ["DCC_MCP_MAYA_SKILL_PATHS"] = "/studio/maya-skills"

from dcc_mcp_core import create_skill_server, McpHttpConfig

# One call: creates registry, dispatcher, SkillCatalog, discovers skills from env vars
server = create_skill_server("maya", McpHttpConfig(port=8765))
handle = server.start()
print(f"Maya MCP server at {handle.mcp_url()}")
# Claude Desktop config: {"mcpServers": {"maya": {"url": "<handle.mcp_url()>"}}}

# On-demand skill discovery workflow (agent-driven):
# 1. tools/list → 6 core tools + __skill__<name> stubs for unloaded skills
#    Core tools: find_skills, list_skills, get_skill_info, load_skill, unload_skill, search_skills
# 2. search_skills(query="bevel") → compact one-line-per-skill results matching search_hint/tool names
# 3. load_skill("maya-bevel") → registers tools + handlers, sends tools/list_changed
# 4. tools/list → new skill tools visible with full schemas
# 5. tools/call maya_bevel__bevel {offset: 0.1} → runs scripts/bevel.py

count = server.discover()               # scan again (optional)
skills = server.list_skills()           # list all with status
server.load_skill("hello-world")        # registers skill tools into ToolRegistry
handle.shutdown()
```

### Full Example: Manual Registry Wiring (low-level)

```python
import os
from pathlib import Path
from dcc_mcp_core import (
    scan_and_load, ToolRegistry, ToolDispatcher,
    McpHttpServer, McpHttpConfig,
    success_result, error_result,
)

os.environ["DCC_MCP_SKILL_PATHS"] = "/opt/maya-skills"
skills, skipped = scan_and_load(dcc_name="maya")

registry = ToolRegistry()
dispatcher = ToolDispatcher(registry)

for skill in skills:
    for script_path in skill.scripts:
        stem = Path(script_path).stem
        action_name = f"{skill.name.replace('-', '_')}__{stem}"
        registry.register(
            name=action_name,
            description=f"[{skill.name}] {skill.description}",
            dcc=skill.dcc, tags=skill.tags, version=skill.version,
        )
        def make_handler(sp):
            def handler(params):
                import subprocess
                proc = subprocess.run(["python", sp], capture_output=True, text=True, timeout=30)
                if proc.returncode == 0:
                    return success_result(proc.stdout.strip(), prompt="Script finished successfully.")
                return error_result("Script failed", proc.stderr.strip())
            return handler
        dispatcher.register_handler(action_name, make_handler(script_path))

server = McpHttpServer(registry, McpHttpConfig(port=0, server_name="maya-mcp"))
handle = server.start()
print(f"MCP server ready: {handle.mcp_url()}")
# Claude Desktop config: {"mcpServers": {"maya": {"url": "<handle.mcp_url()>"}}}
```

---

## Development

- Tool manager: [vx](https://github.com/loonghao/vx)
- Build: `maturin develop --features python-bindings,ext-module`
- Rust tests: `cargo test --workspace` (or `vx just test-rust`)
- Python tests: `vx just test` (requires build first: `vx just dev`)
- Lint: `vx just lint` (clippy + fmt-check + ruff + isort)
- Pre-flight: `vx just preflight` (check + clippy + fmt-check + test-rust)
- Full CI: `vx just ci`
- Release: [Release Please](https://github.com/googleapis/release-please) with Conventional Commits

## AI Agent Tool Priority Guide

When an AI agent needs to interact with DCC software, follow this priority order:

### 1. Skill Discovery (always start here)
```python
# Find relevant skills by keyword
result = search_skills(query="create sphere maya")
# Load the skill to register its tools
load_skill(skill_name="maya-geometry")
# Now the skill's tools appear in tools/list
```

### 2. Skill-Based Tools (preferred over raw API calls)
- Use skill tools (e.g. `maya_geometry__create_sphere`) — they have validated schemas, error handling, and `next-tools` guidance
- Check `ToolAnnotations` for safety hints before calling destructive tools
- Use `next-tools` from tool results to chain follow-up actions

### 3. Diagnostics Tools (for debugging/verification)
```python
diagnostics__screenshot    # verify visual state
diagnostics__audit_log     # check execution history
diagnostics__tool_metrics  # measure performance
diagnostics__process_status # check DCC process health
```

### 4. Direct Registry Access (last resort)
- Only when no skill tool covers the needed operation
- Must validate inputs with `ToolValidator` before execution
- Must use `SandboxPolicy` for AI-initiated calls

### Why Skills First?
1. **Safety**: Skills declare `ToolAnnotations` — agents can check `destructive_hint`, `read_only_hint`
2. **Discoverability**: `search_skills` + `search-hint` keywords find the right tool without trial-and-error
3. **Chainability**: `next-tools` guides follow-up actions, reducing hallucination
4. **Progressive exposure**: Tool groups keep `tools/list` small — agents activate only what they need
5. **Validation**: Skill tools have `input_schema` — parameters are validated before execution

### SKILL.md Description Quality
The `description` field (1-1024 chars) should describe **what the skill does AND when to use it**:
- Bad: `"Helps with geometry."`
- Good: `"Creates and modifies polygon geometry in Maya. Use when user asks to create spheres, cubes, bevel edges, or extrude faces."`

### Progressive Disclosure
Keep `SKILL.md` body < 500 lines / < 5000 tokens; move details to `references/` (loaded on demand by agents).

---

## Related Projects

- [dcc-mcp-rpyc](https://github.com/loonghao/dcc-mcp-rpyc) — RPyC bridge for remote DCC operations
- [dcc-mcp-maya](https://github.com/loonghao/dcc-mcp-maya) — Maya MCP server implementation
