call_meraki_api

Instructions

Implementation Reference

• meraki-mcp-dynamic.py:389-416 (handler)

  The MCP tool handler decorated with @mcp.tool(). Takes section, method, and parameters dict, then delegates to the internal sync function via to_async.

  @mcp.tool()
  async def call_meraki_api(
      section: str,
      method: str,
      parameters: Dict[str, Any] = Field(
          default_factory=dict,
          json_schema_extra={
              'type': 'object',
              'properties': {},
              'additionalProperties': True
          }
      )
  ) -> str:
      """
      Call any Meraki API method - provides access to all 804+ endpoints
  
      Args:
          section: SDK section (organizations, networks, wireless, switch, appliance, camera, devices, sensor, sm, etc.)
          method: Method name (e.g., getOrganizationAdmins, updateNetworkWirelessSsid, getNetworkApplianceFirewallL3FirewallRules)
          parameters: Dict of parameters (e.g., {"networkId": "L_123", "name": "MySSID"})
  
      Examples:
          call_meraki_api(section="organizations", method="getOrganizationAdmins", parameters={"organizationId": "123456"})
          call_meraki_api(section="wireless", method="updateNetworkWirelessSsid", parameters={"networkId": "L_123", "number": "0", "name": "NewSSID", "enabled": True})
          call_meraki_api(section="appliance", method="getNetworkApplianceFirewallL3FirewallRules", parameters={"networkId": "L_123"})
      """
      # Call internal method (parameters is always a dict due to default_factory)
      return await to_async(_call_meraki_method_internal)(section, method, parameters)

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

  Internal async wrapper used by pre-registered tools; delegates 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:416-416 (helper)

  call_meraki_api also delegates to _call_meraki_method_internal via to_async.

  return await to_async(_call_meraki_method_internal)(section, method, parameters)

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

  Core sync function that validates section/method, checks read-only mode, auto-fills org ID, enforces pagination, handles caching, calls the Meraki SDK method, manages large responses, and returns JSON.

  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:389-389 (registration)

  Registration via @mcp.tool() decorator on the call_meraki_api async function.

  @mcp.tool()