Unity MCP插件

一个Python插件,通过stdio传输桥接Semantic Kernel代理与Unity MCP服务器,支持动态工具发现和配置化集成。

  • by Ozymandros
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 Ozymandros/unity-mcp-server-plugin

使用说明

语义内核的Unity MCP插件

Python插件,将语义内核代理与 Unity MCP服务器 通过 stdio传输 (子流程)。

建筑

UnityMCPPlugin
  └── StdioMcpClient          # JSON-RPC 2.0 over stdio
        └── ProcessManager    # asyncio subprocess lifecycle
              └── unity-mcp   # .NET global tool (subprocess)

unity_mcp/
├── exceptions.py      # Full exception hierarchy
├── models.py          # Enums, value objects, protocols (IMcpClient, IProcessManager)
├── security.py        # LogSanitizer, InputValidator
├── process_manager.py # Subprocess lifecycle (start / stop / restart)
├── client.py          # StdioMcpClient — retry, health monitoring, JSON-RPC
├── plugin.py          # UnityMCPPlugin — SK kernel functions, dynamic tool discovery
├── _formatting.py     # format_result, format_tool_list
└── __init__.py        # Public API

SOLID原则适用:

  • 单一职责——每个模块都有一项工作
  • 依赖倒置-- UnityMCPPlugin 接受任何 IMcpClient 通过构造函数注入
  • 打开/关闭——新工具不需要更改代码;它们在运行时被发现

安装

pip install -e .
# dev extras (pytest, mypy, black, flake8)
pip install -e ".[dev]"

从包装好的车轮安装(生产型)

# Build artifacts in dist/
python -m pip install build
python -m build

# Install exact built wheel
pip install dist/unity_mcp_plugin-<version>-py3-none-any.whl

本地主机包+安装自动化(跨平台)

使用辅助脚本自动进行本地打包+安装:

# Full local flow: deps, tests, build, twine check, install wheel into .venv
python scripts/package_install.py

# Fast path (skip tests), recreate both build and target virtualenvs
python scripts/package_install.py --skip-tests --recreate-venvs

# Use a custom target venv
python scripts/package_install.py --venv .venv-local

脚本的作用:

  • 创建构建venv(.venv-build 默认情况下)
  • 安装项目/开发依赖关系和发布工具(build, twine)
  • 运行单元测试(-m "not integration")除非 --skip-tests
  • 将wheel+sdist构建为 dist/
  • 通过以下方式验证工件 twine check
  • 将内置车轮安装到目标venv中(.venv 默认情况下)

先决条件:

  • Python 3.10+
  • semantic-kernel >= 1.0.0
  • unity-mcp .NET全局工具: dotnet tool install -g unity-mcp

快速开始

import asyncio
from unity_mcp import UnityMCPPlugin

async def main():
    plugin = UnityMCPPlugin.create()
    await plugin.initialize()

    result = await plugin.invoke_tool("unity_create_scene", {"path": "Assets/Scenes/Level1.unity"})
    print(result)

    await plugin.cleanup()

asyncio.run(main())

集成示例

该插件需要 unity-mcp 可执行文件在您的环境中可用。 安装一次:

dotnet tool install -g unity-mcp

然后从Python集成:

import asyncio
from unity_mcp import UnityMCPPlugin, UnityMcpOptions

async def main():
    options = UnityMcpOptions(
        executable_path="unity-mcp",  # or absolute path to the executable
        request_timeout_seconds=60,
    )
    plugin = UnityMCPPlugin.create(options)
    await plugin.initialize()
    try:
        tools = await plugin.list_unity_tools()
        print("Discovered tools:")
        print(tools)

        ping_result = await plugin.invoke_tool("ping", {})
        print("Ping:", ping_result)
    finally:
        await plugin.cleanup()

asyncio.run(main())

扩展模式(推荐)——按工具功能

kernel = await UnityMCPPlugin.create_kernel_with_unity()
result = await kernel.invoke("unity", "unity_create_scene", path="Assets/Scenes/Level1.unity")

扩展模式最适合自主代理/规划者,因为每个发现的MCP工具都作为一个单独的SK函数公开,并具有工具级元数据。

路由器模式(向后兼容)——单一通用功能

plugin = UnityMCPPlugin.create()
await plugin.initialize()

# Add only the generic router/list functions
kernel = Kernel()
kernel.add_plugin(plugin, plugin_name="unity")

result = await kernel.invoke(
    "unity",
    "invoke_unity_tool",
    tool_name="unity_create_scene",
    arguments_json='{"path":"Assets/Scenes/Level1.unity"}',
)

路由器模式保持较小的工具定义占用空间(单个通用入口点),但可发现性和工具调用可靠性低于扩展模式。

扩展与路由器的权衡

  • 扩展方式:注册一个插件命名空间(unity 默认情况下),每个MCP工具有一个SK功能;这提高了规划者/自主代理的可发现性和工具调用可靠性。
  • 路由器模式:使工具定义在上下文中保持紧凑(单个通用函数),但代理必须手动推断工具名称/参数,这不太可靠。
  • 推荐:仅当优先考虑最小化工具定义占用空间时,才对代理工作流使用扩展模式和路由器模式。

自定义选项

from unity_mcp import UnityMCPPlugin, UnityMcpOptions, BackoffStrategy

options = UnityMcpOptions(
    executable_path="unity-mcp",
    max_retry_attempts=5,
    backoff_strategy=BackoffStrategy.EXPONENTIAL,
    initial_retry_delay_ms=500,
    request_timeout_seconds=30,
    enable_message_logging=True,
)
plugin = UnityMCPPlugin.create(options)

依赖注入(测试)

from unity_mcp import UnityMCPPlugin, McpResponse

class FakeMcpClient:
    async def connect(self, cancellation_token=None): ...
    async def list_tools(self, cancellation_token=None): return []
    async def invoke_tool(self, tool_name, parameters, cancellation_token=None):
        return McpResponse(id="1", success=True, result={"ok": True})
    async def ping(self, cancellation_token=None): return True
    def is_healthy(self): return True
    async def close(self): ...

plugin = UnityMCPPlugin(client=FakeMcpClient())

主要特点

  • 动态工具发现 --工具在运行时通过以下方式发现 list_tools();没有硬编码包装
  • 确定性注册 --发现的工具在暴露于SK之前按名称排序
  • 使用回退重试 --瞬态故障的可配置线性或指数回退
  • 健康监测 --周期性ping循环; is_healthy() 反映连接状态
  • 安全 — LogSanitizer 从日志中编辑秘密; InputValidator 验证所有工具调用
  • 向后兼容 — UnityMCPClient 为现有代码保留别名

公共API

from unity_mcp import (
    # Plugin
    UnityMCPPlugin,
    # Client
    StdioMcpClient,
    UnityMCPClient,       # backward-compat alias for StdioMcpClient
    # Protocols
    IMcpClient,
    IProcessManager,
    # Configuration
    UnityMcpOptions,
    BackoffStrategy,
    # State enums
    ConnectionState,
    ProcessState,
    # Models
    McpToolDefinition,
    McpParameterDefinition,
    McpReturnType,
    McpRequest,
    McpResponse,
    McpError,
    ProcessInfo,
    # Security
    LogSanitizer,
    InputValidator,
    # Exceptions
    UnityMcpException,
    NetworkException,
    TimeoutException,
    ProtocolException,
    McpServerException,
    ProcessException,
    ConfigurationException,
    TypeConversionException,
)

UnityMcpOptions字段

字段 类型 默认值 描述
executable_path str "unity-mcp" unity mcp可执行文件的路径或名称
connection_timeout_seconds int 30 初始进程启动超时
request_timeout_seconds int 60 每次请求读取超时
max_retry_attempts int 3 瞬态故障重试(0=无重试)
backoff_strategy BackoffStrategy EXPONENTIAL LINEAR 或 EXPONENTIAL
initial_retry_delay_ms int 1000 第一次重试的基本延迟
max_idle_time_seconds int 300 之前的最大空闲时间 is_healthy() 返回False
enable_message_logging bool False 在DEBUG中记录经过清理的请求/响应有效负载
tool_definitions_path str | None None 静态工具定义文件的可选路径

参数类型

McpParameterDefinition.type 遵循JSON模式约定: "string", "number", "integer", "boolean", "object", "array"

元数据保真度注释

扩展模式将MCP参数元数据传播到SK函数元数据中,包括:

  • 精确的参数名称
  • 描述
  • 必需vs可选
  • 默认值(由MCP模式提供时)
  • 类似JSON模式的类型映射(string, number, integer, boolean, array, object)

当前的Python语义内核API并没有完全保留每个JSON模式构造(例如, oneOf、具有完整约束的嵌套对象模式和所有模式关键字)作为一级元数据字段。此插件通过转发当前支持的最丰富的元数据 KernelParameterMetadata (type, description, default_value, is_required,以及 schema_data).

异常层次结构

UnityMcpException
├── NetworkException        — transport / pipe failure
├── TimeoutException        — request exceeded timeout
├── ProtocolException       — malformed JSON-RPC message
├── McpServerException      — server returned a JSON-RPC error
├── ProcessException        — subprocess start/stop failure
├── ConfigurationException  — invalid UnityMcpOptions
└── TypeConversionException — parameter type mismatch

测试

# Unit tests (no server needed)
pytest test_plugin.py -v -m "not integration"

# Integration tests (requires running unity-mcp)
pytest test_plugin.py -v -m integration

更新日志

v3.0.0

  • 霹雳舞:从TCP迁移到 stdio传输 (子流程)
  • 霹雳舞: UnityMCPPlugin.create() 不再需要 host/port;接受 UnityMcpOptions
  • 新: ProcessManager --asyncio子流程生命周期管理
  • 新: StdioMcpClient --基于stdio的JSON-RPC 2.0,具有重试和健康监控功能
  • 新: LogSanitizer + InputValidator 安全层
  • 新:完整的异常层次结构(NetworkException, TimeoutException, ProtocolException等等)
  • 新: UnityMcpOptions 带验证的配置数据类
  • 新:动态工具发现--没有硬编码包装器
  • 新: create_kernel_with_unity() 静态工厂将每个工具注册为自己的工具 KernelFunction
  • 新: invoke_unity_tool + list_unity_tools 通用内核函数
  • 新: McpReturnType 刀具返回值模式模型

v2.0.0版本

  • 将整体重构为 unity_mcp/ 包裹
  • 添加 IMCPClient 依赖注入协议
  • 添加 UnityMCPPlugin.create() 工厂
  • 22个硬编码的工具包装

v1.0.0

  • TCP传输的初始版本

许可证

麻省理工学院