getOrganizationNetworks

Instructions

Implementation Reference

• meraki-mcp-dynamic.py:435-441 (handler)

  The MCP tool handler for 'getOrganizationNetworks'. It is a pre-registered async function decorated with @mcp.tool() that accepts an optional organizationId parameter, builds a params dict, and delegates to call_meraki_method('organizations', 'getOrganizationNetworks', **params).

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

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

  The generic internal API caller (_call_meraki_method_internal) and its async wrapper (call_meraki_method) that actually executes the Meraki SDK call. The handler calls call_meraki_method which invokes this, which does section/method validation, auto-fills organizationId, caching, pagination limits, and finally calls the Meraki SDK's dashboard.organizations.getOrganizationNetworks(**params).

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

  Reference to 'getOrganizationNetworks' in the list of pre_registered_tools within the get_mcp_config tool, confirming it is one of 12 pre-registered convenience tools.

  @mcp.tool()
  async def get_mcp_config() -> str:
      """Get MCP configuration"""
      return json.dumps({
          "mode": "hybrid",
          "description": "12 pre-registered tools + call_meraki_api for full API access",
          "pre_registered_tools": ["getOrganizations", "getOrganizationAdmins", "getOrganizationNetworks",
                                    "getOrganizationDevices", "getNetwork", "getNetworkClients",
                                    "getNetworkEvents", "getNetworkDevices", "getDevice",
                                    "getNetworkWirelessSsids", "getDeviceSwitchPorts", "updateDeviceSwitchPort"],
          "generic_caller": "call_meraki_api - access all 804+ methods",
          "total_available_methods": "804+",
          "read_only_mode": READ_ONLY_MODE,
          "caching_enabled": ENABLE_CACHING,
          "cache_ttl_seconds": CACHE_TTL_SECONDS,
          "file_caching_enabled": ENABLE_FILE_CACHING,
          "max_response_tokens": MAX_RESPONSE_TOKENS,
          "max_per_page": MAX_PER_PAGE,
          "response_cache_dir": RESPONSE_CACHE_DIR,
          "organization_id_configured": bool(MERAKI_ORG_ID),
          "api_key_configured": bool(MERAKI_API_KEY)
      }, indent=2)

• meraki-mcp.py:68-80 (helper)

  In the second implementation (meraki-mcp.py), async_get_organization_networks is created by wrapping dashboard.organizations.getOrganizationNetworks with to_async() on line 71.

  # Create async versions of commonly used Meraki API methods
  async_get_organizations = to_async(dashboard.organizations.getOrganizations)
  async_get_organization = to_async(dashboard.organizations.getOrganization)
  async_get_organization_networks = to_async(dashboard.organizations.getOrganizationNetworks)
  async_get_organization_devices = to_async(dashboard.organizations.getOrganizationDevices)
  async_get_network = to_async(dashboard.networks.getNetwork)
  async_get_network_devices = to_async(dashboard.networks.getNetworkDevices)
  async_get_network_clients = to_async(dashboard.networks.getNetworkClients)
  async_get_device = to_async(dashboard.devices.getDevice)
  async_update_device = to_async(dashboard.devices.updateDevice)
  async_get_wireless_ssids = to_async(dashboard.wireless.getNetworkWirelessSsids)
  async_update_wireless_ssid = to_async(dashboard.wireless.updateNetworkWirelessSsid)

• meraki-mcp.py:217-223 (handler)

  The handler in meraki-mcp.py for the tool named 'get_networks' (note: different tool name) that calls async_get_organization_networks(organization_id) to get organization networks.

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