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, }