语义内核的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 APISOLID原则适用:
- 单一职责——每个模块都有一项工作
- 依赖倒置--
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--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.0unity-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传输的初始版本
许可证
麻省理工学院
