Moku MCP Server

Instructions

Implementation Reference

• src/moku_mcp/server.py:151-241 (handler)

  The attach_moku method implementation in MokuServer, responsible for connecting to a Moku device and handling ownership.

  async def attach_moku(self, device_id: str, force: bool = False):
      """
      Connect to Moku device and assume ownership.
  
      Args:
          device_id: IP address, name, or serial number
          force: Force connection even if owned by another client
  
      Returns:
          {
              "status": "connected",
              "device": {
                  "ip": "192.168.1.100",
                  "name": "Lilo",
                  "serial": "MG106B",
                  "platform": "Moku:Go"
              }
          }
  
      Implementation: See IMPLEMENTATION_GUIDE.md Section 3.2
      """
      from moku.instruments import MultiInstrument
      from .utils import resolve_device_identifier, load_device_cache
  
      # Check if already connected
      if self.moku_instance:
          if self.connected_device == device_id:
              return {
                  "status": "already_connected",
                  "message": f"Already connected to {device_id}",
                  "device": {"ip": self.connected_device, "platform": "Moku:Go"},
              }
          else:
              return {
                  "status": "error",
                  "message": f"Already connected to {self.connected_device}. Release first.",
                  "suggestion": "Call release_moku() before connecting to a different device",
              }
  
      # Resolve device_id to IP
      ip = resolve_device_identifier(device_id)
      if not ip:
          # If not in cache, check if it's a valid IP
          if "." in device_id and device_id.replace(".", "").isdigit():
              ip = device_id
          else:
              return {
                  "status": "error",
                  "message": f"Device '{device_id}' not found in cache",
                  "suggestion": "Run discover_mokus() first to find devices",
              }
  
      # Try to connect (platform_id=2 for Moku:Go)
      try:
          logger.info(f"Attempting to connect to {ip} (force={force})")
          self.moku_instance = MultiInstrument(ip, platform_id=2, force_connect=force)
          self.connected_device = ip
  
          # Get device info from cache
          cache = load_device_cache()
          device_info = cache.find_by_ip(ip)
  
          logger.info(f"Successfully connected to Moku at {ip}")
  
          return {
              "status": "connected",
              "device": {
                  "ip": ip,
                  "name": device_info.canonical_name if device_info else "Unknown",
                  "serial": device_info.serial_number if device_info else "Unknown",
                  "platform": "Moku:Go",
              },
          }
  
      except ConnectionError as e:
          logger.error(f"Connection failed: {e}")
          return {
              "status": "error",
              "message": f"Could not connect to {ip}. Device may be offline or owned by another client.",
              "suggestion": "Try with force=True to take ownership, or wait for current owner to disconnect.",
              "details": str(e),
          }
      except Exception as e:
          logger.error(f"Unexpected error connecting to {ip}: {e}")
          self.moku_instance = None
          self.connected_device = None
          return {
              "status": "error",
              "message": f"Failed to connect to {ip}",
              "details": str(e),
          }