getOrganizationDevices

Instructions

Implementation Reference

• meraki-mcp-dynamic.py:443-449 (handler)

  MCP tool handler for getOrganizationDevices - decorated with @mcp.tool(), accepts optional organizationId, builds params dict, delegates to call_meraki_method which routes to the Meraki SDK's organizations.getOrganizationDevices

  @mcp.tool()
  async def getOrganizationDevices(organizationId: str = None) -> str:
      """Get organization devices"""
      params = {}
      if organizationId:
          params['organizationId'] = organizationId
      return await call_meraki_method("organizations", "getOrganizationDevices", **params)

• meraki-mcp.py:225-231 (handler)

  Alternate MCP tool handler named get_devices in the simpler MCP implementation. Uses an async wrapper around dashboard.organizations.getOrganizationDevices and calls it with the organization ID.

  # Get devices from Meraki
  @mcp.tool()
  async def get_devices(org_id: str = None) -> str:
      """Get a list of devices from Meraki"""
      organization_id = org_id or MERAKI_ORG_ID
      devices = await async_get_organization_devices(organization_id)
      return json.dumps(devices, indent=2)

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

  The call_meraki_method async wrapper that getOrganizationDevices delegates to. Converts the synchronous Meraki SDK call to async via to_async and routes to _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-365 (helper)

  Internal synchronous implementation that actually calls the Meraki SDK method. Validates sections/methods, auto-fills org ID, handles caching, enforces pagination limits, and manages large response truncation.

  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)

• meraki-mcp-dynamic.py:655-658 (registration)

  Registration reference: 'getOrganizationDevices' is listed in the pre_registered_tools array within the get_mcp_config tool, confirming it is one of the 12 pre-registered convenience tools.

  "pre_registered_tools": ["getOrganizations", "getOrganizationAdmins", "getOrganizationNetworks",
                            "getOrganizationDevices", "getNetwork", "getNetworkClients",
                            "getNetworkEvents", "getNetworkDevices", "getDevice",
                            "getNetworkWirelessSsids", "getDeviceSwitchPorts", "updateDeviceSwitchPort"],

• meraki-mcp.py:72-72 (helper)

  Async wrapper creation for dashboard.organizations.getOrganizationDevices - wraps the synchronous SDK method with to_async() so it can be called asynchronously.

  async_get_organization_devices = to_async(dashboard.organizations.getOrganizationDevices)