Revit MCP Server

Instructions

Implementation Reference

• xml_revit_mcp/server.py:41-44 (registration)

  The create_levels tool is included in the ARCHITECTURAL_TOOLS list, which groups tools for registration with the MCP server.

  ARCHITECTURAL_TOOLS = [
      create_levels, create_floor_plan_views, create_grids, create_walls, create_floors,
      create_door_windows, create_rooms, create_room_tags, create_family_instances, create_sheets
  ]

• xml_revit_mcp/server.py:134-147 (registration)

  The register_tools function iterates over ARCHITECTURAL_TOOLS (including create_levels) and registers each tool function using the FastMCP server's tool decorator.

  def register_tools(server: FastMCP) -> None:
      """注册所有工具到MCP服务器"""
      # 注册建筑工具
      for tool in ARCHITECTURAL_TOOLS:
          server.tool()(tool)
  
      # 注册MEP工具
      for tool in MEP_TOOLS:
          server.tool()(tool)
  
      # 注册通用工具
      for tool in GENERAL_TOOLS:
          server.tool()(tool)

• xml_revit_mcp/server.py:248-248 (registration)

  Explicit call to register_tools(mcp), which registers the create_levels tool among others on the MCP server instance.

  register_tools(mcp)

• xml_revit_mcp/server.py:58-94 (helper)

  Utility function to get or create a persistent RevitConnection, which tool handlers like create_levels would use to send commands to the Revit plugin.

  def get_revit_connection() -> RevitConnection:
      """
      获取或创建持久的Revit连接
  
      返回:
          RevitConnection: 与Revit的连接对象
  
      异常:
          Exception: 连接失败时抛出
      """
      global _connection, _enabled
  
      if _connection is not None:
          try:
              # 测试连接是否有效
              result = _connection.send_command("get_polyhaven_status")
              _enabled = result.get("enabled", False)
              return _connection
          except Exception as e:
              logger.warning(f"现有连接已失效: {str(e)}")
              try:
                  _connection.disconnect()
              except:
                  pass
              _connection = None
  
      # 创建新连接
      if _connection is None:
          _connection = RevitConnection(host="localhost", port=_port)
          if not _connection.connect():
              logger.error("无法连接到Revit")
              _connection = None
              raise Exception(
                  "无法连接到Revit。请确保Revit插件正在运行。")
          logger.info("已创建新的持久连接到Revit")
  
      return _connection

• xml_revit_mcp/revit_connection.py:175-241 (helper)

  Core method in RevitConnection that sends JSON-RPC commands (e.g., 'CreateLevels') to the Revit plugin; tool handlers would call this with command_type='create_levels' or similar.

  def send_command(self, command_type: str, params: Union[Dict[str, Any], List[Dict[str, Any]]] = None) -> Dict[
      str, Any]:
      """
      向Revit发送命令并返回响应
  
      参数:
          command_type (str): 命令类型
          params (Dict[str, Any] 或 List[Dict[str, Any]]): 命令参数
  
      返回:
          Dict[str, Any]: 命令响应
  
      异常:
          ConnectionError: 连接错误
          TimeoutError: 请求超时
          ValueError: 参数或响应无效
          Exception: 其他错误
      """
      # 确保连接
      if not self.sock and not self.connect():
          raise ConnectionError("无法连接到Revit")
  
      try:
          logger.info(f"发送命令: {command_type}")
          logger.debug(f"命令参数: {params}")
  
          # 导入并创建请求对象
          from .rpc import JsonRPCRequest, JsonRPCResponse
          command = JsonRPCRequest(method=command_type, params=params)
          command_json = json.dumps(command.__dict__)
  
          # 发送命令
          self.sock.sendall(command_json.encode('utf-8'))
          logger.debug("命令已发送，等待响应...")
  
          # 使用 receive_full_response 接收完整数据流
          response_data = self.receive_full_response()
          logger.debug(f"已接收 {len(response_data)} 字节数据")
  
          # 解析响应
          try:
              response_dict = json.loads(response_data.decode('utf-8'))
              response = JsonRPCResponse(
                  id=response_dict.get("id"),
                  result=response_dict.get("result"),
                  error=response_dict.get("error")
              )
          except json.JSONDecodeError as e:
              logger.error(f"无法解析Revit响应: {str(e)}")
              if response_data:
                  logger.error(f"原始响应 (前200字节): {response_data[:200]}")
              raise ValueError(f"无效的Revit响应: {str(e)}")
  
          # 处理错误
          if response.error:
              error_message = response.error.get("message", "未知错误")
              error_code = response.error.get("code", -1)
              error_data = response.error.get("data")
  
              logger.error(f"Revit错误 (代码: {error_code}): {error_message}")
              if error_data:
                  logger.error(f"错误数据: {error_data}")
  
              raise Exception(f"Revit错误: {error_message}")
  
          return response.result or {}