ms-sentinel-mcp-server

Instructions

Implementation Reference

• tools/threat_intel_tools.py:100-186 (handler)

  The async run method implements the tool's core logic: extracts 'ip' parameter, initializes Azure SecurityInsights client, calls ip_geodata.get via run_in_thread, processes response to dict, handles errors.

  async def run(self, ctx: Context, **kwargs):
      """
      Get geolocation data for an IP address.
  
      Args:
          ctx (Context): The MCP tool context.
          **kwargs: IP address as 'ip' parameter.
  
      Returns:
          dict: Results as described in the class docstring.
      """
  
      # Extract parameters
      # Extract ip parameter using the centralized parameter extraction from MCPToolBase
      ip = self._extract_param(kwargs, "ip")
  
      if not ip:
          return {"error": "ip parameter is required", "valid": False}
  
      # Get Azure context
      workspace_name, resource_group, subscription_id = self.get_azure_context(ctx)
  
      # Get security insights client
      client = None
      try:
          client = self.get_securityinsight_client(subscription_id)
      except Exception as e:
          self.logger.error("Error initializing Azure SecurityInsights client: %s", e)
          return {
              "error": (
                  f"Azure SecurityInsights client initialization failed: {str(e)}"
              ),
              "valid": False,
          }
  
      if client is None:
          return {
              "error": "Azure SecurityInsights client is not initialized",
              "valid": False,
          }
  
      # Validate Azure context
      valid = self.validate_azure_context(
          client is not None,
          workspace_name,
          resource_group,
          subscription_id,
          self.logger,
      )
      if not valid:
          return {
              "error": "Missing required Azure context or SDK components",
              "valid": False,
          }
  
      try:
          # Get geolocation data for the IP address
          # Based on SDK testing, ip_geodata.get() doesn't accept workspace_name
          geodata = await run_in_thread(
              client.ip_geodata.get,
              resource_group_name=resource_group,
              ip_address=ip,
          )
  
          # Process geodata result
          # Return the full geodata object
          geodata_dict = {}
          if hasattr(geodata, "as_dict"):
              geodata_dict = geodata.as_dict()
          else:
              # If as_dict() is not available, try to convert to dict directly
              geodata_dict = dict(geodata) if geodata else {}
  
          # Ensure we have at least the IP in the response
          if not geodata_dict or not geodata_dict.get("ip"):
              geodata_dict["ip"] = ip
  
          return {
              "geodata": geodata_dict,
              "valid": True,
          }
      except Exception as e:
          self.logger.error("Error retrieving IP geodata for %s: %s", ip, e)
          return {
              "error": f"Error retrieving IP geodata for {ip}: {str(e)}",
              "valid": False,
          }

• tools/threat_intel_tools.py:97-99 (schema)

  Tool name and description defining the schema and expected input ('ip' parameter) and output format as described in class docstring.

  name = "sentinel_ip_geodata_get"
  description = "Get geolocation data for an IP address"

• tools/threat_intel_tools.py:438-438 (registration)

  Registers the SentinelIPGeodataGetTool with the MCP instance in the register_tools function.

  SentinelIPGeodataGetTool.register(mcp)

• tools/threat_intel_tools.py:85-187 (handler)

  Full class definition of the MCPToolBase subclass implementing the sentinel_ip_geodata_get tool, including docstring for schema.

  class SentinelIPGeodataGetTool(MCPToolBase):
      """
      Tool to get geolocation data for an IP address.
  
      Returns:
          dict: {
              'geodata': dict,   # Geolocation data as returned by the API
              'valid': bool,     # True if successful
              'error': str (optional)
          }
      """
  
      name = "sentinel_ip_geodata_get"
      description = "Get geolocation data for an IP address"
  
      async def run(self, ctx: Context, **kwargs):
          """
          Get geolocation data for an IP address.
  
          Args:
              ctx (Context): The MCP tool context.
              **kwargs: IP address as 'ip' parameter.
  
          Returns:
              dict: Results as described in the class docstring.
          """
  
          # Extract parameters
          # Extract ip parameter using the centralized parameter extraction from MCPToolBase
          ip = self._extract_param(kwargs, "ip")
  
          if not ip:
              return {"error": "ip parameter is required", "valid": False}
  
          # Get Azure context
          workspace_name, resource_group, subscription_id = self.get_azure_context(ctx)
  
          # Get security insights client
          client = None
          try:
              client = self.get_securityinsight_client(subscription_id)
          except Exception as e:
              self.logger.error("Error initializing Azure SecurityInsights client: %s", e)
              return {
                  "error": (
                      f"Azure SecurityInsights client initialization failed: {str(e)}"
                  ),
                  "valid": False,
              }
  
          if client is None:
              return {
                  "error": "Azure SecurityInsights client is not initialized",
                  "valid": False,
              }
  
          # Validate Azure context
          valid = self.validate_azure_context(
              client is not None,
              workspace_name,
              resource_group,
              subscription_id,
              self.logger,
          )
          if not valid:
              return {
                  "error": "Missing required Azure context or SDK components",
                  "valid": False,
              }
  
          try:
              # Get geolocation data for the IP address
              # Based on SDK testing, ip_geodata.get() doesn't accept workspace_name
              geodata = await run_in_thread(
                  client.ip_geodata.get,
                  resource_group_name=resource_group,
                  ip_address=ip,
              )
  
              # Process geodata result
              # Return the full geodata object
              geodata_dict = {}
              if hasattr(geodata, "as_dict"):
                  geodata_dict = geodata.as_dict()
              else:
                  # If as_dict() is not available, try to convert to dict directly
                  geodata_dict = dict(geodata) if geodata else {}
  
              # Ensure we have at least the IP in the response
              if not geodata_dict or not geodata_dict.get("ip"):
                  geodata_dict["ip"] = ip
  
              return {
                  "geodata": geodata_dict,
                  "valid": True,
              }
          except Exception as e:
              self.logger.error("Error retrieving IP geodata for %s: %s", ip, e)
              return {
                  "error": f"Error retrieving IP geodata for {ip}: {str(e)}",
                  "valid": False,
              }