Moku MCP Server

Instructions

Implementation Reference

• src/moku_mcp/server.py:289-413 (handler)

  The push_config method handles the deployment of MokuConfig to the connected device by configuring instrument slots and setting up signal routing.

  async def push_config(self, config_dict: dict):
      """
      Deploy MokuConfig to connected device.
  
      Args:
          config_dict: MokuConfig serialized as dict
  
      Returns:
          {
              "status": "deployed",
              "slots_configured": [1, 2],
              "routing_configured": True
          }
  
      Implementation: See IMPLEMENTATION_GUIDE.md Section 3.4
      """
      from moku.instruments import CloudCompile, Oscilloscope
      from pydantic import ValidationError
  
      if not self.moku_instance:
          return {
              "status": "error",
              "message": "Not connected to any device",
              "suggestion": "Call attach_moku first",
          }
  
      # Validate and parse config
      try:
          config = MokuConfig.model_validate(config_dict)
      except ValidationError as e:
          logger.error(f"Invalid config: {e}")
          return {"status": "error", "message": "Invalid MokuConfig", "errors": e.errors()}
  
      # Validate routing
      errors = config.validate_routing()
      if errors:
          return {
              "status": "error",
              "message": "Invalid routing configuration",
              "errors": errors,
          }
  
      deployed_slots = []
  
      # Deploy instruments to slots
      for slot_num, slot_config in config.slots.items():
          try:
              if slot_config.instrument == "CloudCompile":
                  if not slot_config.bitstream:
                      logger.warning(f"Slot {slot_num}: No bitstream specified")
                      continue
  
                  logger.info(f"Deploying CloudCompile to slot {slot_num}")
                  self.moku_instance.set_instrument(
                      slot_num, CloudCompile, bitstream=slot_config.bitstream
                  )
  
                  # Apply control registers if specified
                  if slot_config.control_registers:
                      cc = self.moku_instance.get_instrument(slot_num)
                      for reg, value in slot_config.control_registers.items():
                          cc.write_register(reg, value)
                          logger.debug(f"Slot {slot_num}: Set register {reg} = {value:#x}")
  
                  deployed_slots.append(slot_num)
                  logger.info(f"Successfully deployed CloudCompile to slot {slot_num}")
  
              elif slot_config.instrument == "Oscilloscope":
                  logger.info(f"Deploying Oscilloscope to slot {slot_num}")
                  osc = self.moku_instance.set_instrument(slot_num, Oscilloscope)
  
                  # Apply settings if specified
                  if slot_config.settings and "timebase" in slot_config.settings:
                      osc.set_timebase(*slot_config.settings["timebase"])
                      logger.debug(f"Slot {slot_num}: Set timebase {slot_config.settings['timebase']}")
  
                  deployed_slots.append(slot_num)
                  logger.info(f"Successfully deployed Oscilloscope to slot {slot_num}")
  
              else:
                  logger.warning(
                      f"Slot {slot_num}: Instrument '{slot_config.instrument}' not supported yet"
                  )
  
          except Exception as e:
              logger.error(f"Failed to deploy slot {slot_num}: {e}")
              return {
                  "status": "error",
                  "message": f"Failed to deploy instrument to slot {slot_num}",
                  "details": str(e),
                  "slots_deployed": deployed_slots,
              }
  
      # Configure routing
      routing_configured = False
      if config.routing:
          try:
              # Convert routing to dict format expected by Moku API
              connections = []
              for conn in config.routing:
                  connections.append(
                      {"source": conn.source, "destination": conn.destination}
                  )
  
              self.moku_instance.set_connections(connections)
              routing_configured = True
              logger.info(f"Configured {len(connections)} routing connections")
  
          except Exception as e:
              logger.error(f"Failed to configure routing: {e}")
              return {
                  "status": "partial_success",
                  "message": "Instruments deployed but routing failed",
                  "slots_configured": deployed_slots,
                  "routing_error": str(e),
              }
  
      # Cache the config for get_config()
      self.last_config = config
  
      return {
          "status": "deployed",
          "slots_configured": deployed_slots,
          "routing_configured": routing_configured,
      }

• src/moku_mcp/tools.py:148-157 (registration)

  The tools.py file registers and dispatches the push_config tool call to the server's push_config method.

  elif name == "push_config":
      if not server.moku_instance:
          error = {
              "status": "error",
              "message": "Not connected to any device",
              "suggestion": "Call attach_moku first",
          }
          return [TextContent(type="text", text=json.dumps(error, indent=2))]
  
      result = await server.push_config(**arguments)