Follow these steps to get started with professional threat intelligence analysis
1
Submit IPs
Upload your IP addresses of interest through our secure interface. Our platform handles datasets up to 5000 addresses.
2
Analysis
Our deterministic ensemble pipeline analyzes patterns, identifies relationships, and generates threat intelligence automatically — same input, same clusters, same reasoning, every run.
3
Receive reports
Get comprehensive threat intelligence reports with IOCs, YARA rules, and hunting queries.
4
Execute hunting queries
Use our automated hunting query execution service to validate findings and monitor for new threats.
User Guide
API Usage Guide
API Usage Guide
ClusterHawk provides a public API that allows you to programmatically submit prediction jobs and retrieve results using your trained models. This guide covers everything you need to know to integrate ClusterHawk into your security workflows.
Prerequisites
Before using the API, you need:
Team tier or higher subscription - API access is not available on Analyst plans
At least one trained model - You must have completed a training job to create models for predictions
An active API key - Generate one from your Profile page (shown for 30 seconds only)
Generating an API Key
To generate an API key:
Navigate to your Profile page
Scroll to the "API Access" section
Click "Generate API Key" (requires Team tier or higher)
Copy your API key immediately - it's only shown for 30 seconds
Store the key securely in your application's configuration
Authentication
All API requests must include your API key for authentication. You can provide the API key in two ways:
Retrieve the prediction results once the job is completed. For user-trained models, each prediction includes a label field with the actor label from the training job.
Each prediction row carries the contract's trust gate in the kind field. Read it first. confident_match rows are the actionable subset. ambiguous_diffuse and ambiguous_split rows hedge across multiple candidates; iterate the candidates array as a unit rather than relying on top-1. out_of_distribution rows do not match any trained pattern. predicted_cluster and confidence are intentionally null on those rows to prevent silent false positives in SIEM joins. top1_minus_top2 and effective_n are diagnostic fields that quantify the gap between the leading candidates and the entropy-effective number of candidates respectively.
Secure API Key Storage: Store your API key securely using environment variables or secure configuration management. Never commit API keys to version control.
Error Handling: Implement proper error handling for all API responses, including authentication, rate limiting, and server errors.
Polling for Results: When checking job status, implement exponential backoff to avoid excessive polling. Start with 5-second intervals and increase gradually.
Batch Processing: Submit multiple IP addresses in a single prediction job rather than making individual requests for each IP.
Monitor Quotas: Track your prediction usage through the Dashboard to avoid hitting monthly limits unexpectedly.
Model Management: Use descriptive model names that clearly indicate their purpose and training data characteristics.
Python Example Integration
Here's a complete Python example for integrating with the ClusterHawk API:
import requests
import time
import os
class ClusterHawkAPI:
def __init__(self, api_key, base_url="https://clusterhawk.chawkr.com"):
self.api_key = api_key
self.base_url = base_url
self.headers = {
"X-API-Key": api_key,
"Content-Type": "application/json"
}
def submit_prediction(self, model_name, ip_addresses, job_name=None):
"""Submit a prediction job."""
payload = {
"model_name": model_name,
"ip_addresses": ip_addresses,
"job_name": job_name or f"Prediction-{int(time.time())}"
}
response = requests.post(
f"{self.base_url}/api/v1/public/predict",
headers=self.headers,
json=payload
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
def get_job_status(self, job_id):
"""Check job status."""
response = requests.get(
f"{self.base_url}/api/v1/public/jobs/{job_id}/status",
headers=self.headers
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
def get_results(self, job_id):
"""Get prediction results."""
response = requests.get(
f"{self.base_url}/api/v1/public/jobs/{job_id}/results",
headers=self.headers
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
def list_models(self):
"""List available models."""
response = requests.get(
f"{self.base_url}/api/v1/public/models",
headers=self.headers
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
def check_quota(self):
"""Check concurrent job quota availability."""
response = requests.get(
f"{self.base_url}/api/v1/public/quota/concurrent-jobs",
headers=self.headers
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
def predict_and_wait(self, model_name, ip_addresses, job_name=None, max_wait=300):
"""Submit prediction and wait for completion."""
# Submit job
job_info = self.submit_prediction(model_name, ip_addresses, job_name)
job_id = job_info["job_id"]
print(f"Job submitted: {job_id}")
# Wait for completion
start_time = time.time()
while time.time() - start_time < max_wait:
status = self.get_job_status(job_id)
print(f"Job status: {status['status']}")
if status["status"] == "completed":
return self.get_results(job_id)
elif status["status"] in ["failed", "cancelled"]:
raise Exception(f"Job {status['status']}")
time.sleep(10) # Wait 10 seconds before next check
raise Exception("Job timeout")
# Usage example
def main():
api_key = os.getenv("CHAWKR_API_KEY")
if not api_key:
raise Exception("Please set CHAWKR_API_KEY environment variable")
client = ClusterHawkAPI(api_key)
# Check quota before submitting
quota_info = client.check_quota()
print(f"Quota Status: {quota_info['quota']['current_usage']}/{quota_info['quota']['max_concurrent']} jobs used")
print(f"Available slots: {quota_info['quota']['available']}")
if not quota_info['quota']['can_submit']:
print("Cannot submit job - quota limit reached")
return
# List available models first
models = client.list_models()
print(f"Available models: {[m['model_name'] for m in models['models']]}")
# Predict using your trained model
results = client.predict_and_wait(
model_name="network-classification-v1",
ip_addresses=["192.168.1.100", "10.0.0.50", "203.0.113.42"],
job_name="Security Assessment"
)
# Process results
prediction_data = results['results']['prediction']
print(f"Analyzed {prediction_data['total_predictions']} IP addresses")
for prediction in prediction_data["predictions"]:
ip = prediction["ip"]
kind = prediction.get("kind", "unknown")
cluster = prediction.get("predicted_cluster")
confidence = prediction.get("confidence")
candidates = prediction.get("candidates", [])
# Read the contract's kind field first — it is the trust gate.
# OOD rows have predicted_cluster=null; act on behavioral evidence,
# not on the cluster attribution.
if kind == "out_of_distribution":
print(f"IP {ip}: out-of-distribution (no fingerprint match)")
elif kind == "confident_match":
print(f"IP {ip}: Cluster {cluster} (confident, confidence={confidence:.2f})")
else:
# ambiguous_diffuse / ambiguous_split — inspect the candidate set
cand_str = ", ".join(
f"{c['cluster_id']}:{c['confidence']:.2f}" for c in candidates
)
print(f"IP {ip}: Cluster {cluster} ({kind}, candidates=[{cand_str}])")
if __name__ == "__main__":
main()
Support and Troubleshooting
If you encounter issues with the API:
Check your subscription tier - API access requires Team tier or higher
Verify your API key - Ensure it's correctly formatted and not expired
Confirm model availability - The model name must match exactly (case-sensitive)
Monitor quota usage - Check the Dashboard for remaining prediction quota
We use essential cookies to provide core functionality for authentication and payment processing. These cookies are necessary for the website to function properly and cannot be disabled. We do not use any non-essential cookies for analytics or tracking. For more information, please read our Privacy Policy.
