getDeviceSwitchPorts

Instructions

Implementation Reference

• meraki-mcp-dynamic.py:485-488 (handler)

  The MCP tool handler for 'getDeviceSwitchPorts'. It is registered with @mcp.tool() decorator and delegates to call_meraki_method with section='switch' and method='getDeviceSwitchPorts', passing the serial parameter.

  @mcp.tool()
  async def getDeviceSwitchPorts(serial: str) -> str:
      """Get switch ports for a device"""
      return await call_meraki_method("switch", "getDeviceSwitchPorts", serial=serial)

• meraki-mcp-dynamic.py:485-488 (registration)

  The @mcp.tool() decorator on line 485 registers 'getDeviceSwitchPorts' as an MCP tool. It is also listed in the pre_registered_tools array in get_mcp_config() (line 658).

  @mcp.tool()
  async def getDeviceSwitchPorts(serial: str) -> str:
      """Get switch ports for a device"""
      return await call_meraki_method("switch", "getDeviceSwitchPorts", serial=serial)

• meraki-mcp-dynamic.py:385-387 (helper)

  The call_meraki_method async wrapper that call_meraki_method delegates to to_async(_call_meraki_method_internal).

  async def call_meraki_method(section: str, method: str, **params) -> str:
      """Internal async wrapper for pre-registered tools"""
      return await to_async(_call_meraki_method_internal)(section, method, params)

• meraki-mcp-dynamic.py:264-383 (helper)

  The _call_meraki_method_internal function that handles the actual Meraki API call. It validates the section/method, enforces pagination/read-only mode, handles caching, and invokes the meraki dashboard SDK method (dashboard.switch.getDeviceSwitchPorts(serial)).

  def _call_meraki_method_internal(section: str, method: str, params: dict) -> str:
      """Internal helper to call Meraki API methods"""
      pagination_limited = False
      original_params = params.copy()
  
      try:
          # Validate section
          if not hasattr(dashboard, section):
              return json.dumps({
                  "error": f"Invalid section '{section}'",
                  "available_sections": SDK_SECTIONS
              }, indent=2)
  
          section_obj = getattr(dashboard, section)
  
          # Validate method
          if not hasattr(section_obj, method):
              return json.dumps({
                  "error": f"Method '{method}' not found in section '{section}'"
              }, indent=2)
  
          method_func = getattr(section_obj, method)
  
          if not callable(method_func):
              return json.dumps({"error": f"'{method}' is not callable"}, indent=2)
  
          # Determine operation type
          is_read = is_read_only_operation(method)
          is_write = is_write_operation(method)
  
          # Read-only mode check
          if READ_ONLY_MODE and is_write:
              return json.dumps({
                  "error": "Write operation blocked - READ_ONLY_MODE is enabled",
                  "method": method,
                  "hint": "Set READ_ONLY_MODE=false in .env to enable"
              }, indent=2)
  
          # Auto-fill org ID if needed
          sig = inspect.signature(method_func)
          method_params = [p for p in sig.parameters.keys() if p != 'self']
  
          if 'organizationId' in method_params and 'organizationId' not in params and MERAKI_ORG_ID:
              params['organizationId'] = MERAKI_ORG_ID
  
          # Enforce pagination limits
          params_before = params.copy()
          params = enforce_pagination_limits(params, method)
          if params != params_before:
              pagination_limited = True
  
          # Check cache for read operations
          if ENABLE_CACHING and is_read:
              cache_key = create_cache_key(section, method, params)
              cached = cache.get(cache_key)
              if cached is not None:
                  if isinstance(cached, dict):
                      cached['_from_cache'] = True
                  return json.dumps(cached, indent=2)
  
          # Call the method
          result = method_func(**params)
  
          # Invalidate cached read results for this section after any write operation
          if ENABLE_CACHING and is_write:
              cache.invalidate(section)
  
          # Check response size and handle large responses
          result_json = json.dumps(result)
          estimated_tokens = estimate_token_count(result_json)
  
          if ENABLE_FILE_CACHING and estimated_tokens > MAX_RESPONSE_TOKENS:
              # Save full response to file
              filepath = save_response_to_file(result, section, method, original_params)
  
              # Create truncated response with metadata
              truncated_response = create_truncated_response(result, filepath, section, method, original_params)
  
              # Add pagination warning if limits were enforced
              if pagination_limited:
                  truncated_response["_pagination_limited"] = True
                  truncated_response["_pagination_message"] = f"Request modified: pagination limited to {MAX_PER_PAGE} items per page"
  
              # Cache the truncated response (not the full result)
              if ENABLE_CACHING and is_read:
                  cache_key = create_cache_key(section, method, params)
                  cache.set(cache_key, truncated_response)
  
              return json.dumps(truncated_response, indent=2)
  
          # Normal response (small enough)
          response_data = result
          if pagination_limited and isinstance(response_data, dict):
              response_data["_pagination_limited"] = True
              response_data["_pagination_message"] = f"Request modified: pagination limited to {MAX_PER_PAGE} items per page"
  
          # Cache read results
          if ENABLE_CACHING and is_read:
              cache_key = create_cache_key(section, method, params)
              cache.set(cache_key, response_data)
  
          return json.dumps(response_data, indent=2)
  
      except meraki.exceptions.APIError as e:
          return json.dumps({
              "error": "Meraki API Error",
              "message": str(e),
              "status": getattr(e, 'status', 'unknown')
          }, indent=2)
      except TypeError as e:
          return json.dumps({
              "error": "Invalid parameters",
              "message": str(e),
              "hint": f"Use get_method_info(section='{section}', method='{method}') for parameter details"
          }, indent=2)
      except Exception as e:
          return json.dumps({
              "error": str(e),
              "type": type(e).__name__
          }, indent=2)

• meraki-mcp-dynamic.py:486-488 (schema)

  The input schema for getDeviceSwitchPorts: requires a single 'serial' parameter of type str. The return type is str (JSON-encoded result).

  async def getDeviceSwitchPorts(serial: str) -> str:
      """Get switch ports for a device"""
      return await call_meraki_method("switch", "getDeviceSwitchPorts", serial=serial)