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. no_input_features rows exposed nothing the model reads, so confidence is null — but predicted_cluster is retained, because it correctly groups the host with the other input-less hosts. Match these rows on kind: a predicted_cluster is null test will not find them and will route them into your matched branch. Treat them as un-assessed rather than clean, and review the raw scan data. 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.
Each row also carries a per-prediction explanation of why the model placed that IP in its cluster, separate from the cluster's general characteristics. explanation_status is ok, ok:no_signals, partial, unavailable:<reason>, or disabled. A row comes back ok:no_signals when the deciding models were attributed normally but none of the pull landed on a feature we can show you, so its toward and away lists are empty. Test ok as a prefix rather than for equality: every ok state carries a full explanation, and more qualifiers may follow. When present, explanation holds the structured account: a mode that follows the row's kind (single, contrastive, diffuse, none_fits, or no_input_features), the target and reference clusters being compared, and model_votes for the contributing models. Each candidate carries toward / away signals, and every signal is a readable feature: a label, a weight, the observed value, and a value_state of matched, oov (present but unrecognized), or absent. explained_share and interpretable_share report how much of the decision was accounted for and how much of that rests on readable signals; when part of it rested on the model's internal features, the row carries a partly_internal_features caveat.
Signals the model derived internally are withheld from toward and away rather than listed under a placeholder name, so a signal you receive is always one you can go and check against the host. Their weight is not hidden, only their identity: it is what interpretable_share measures the absence of, and what the partly_internal_features caveat flags. How much a model can name depends on where it makes its decision. Core and Neural models work on the fields from your data and name them. Deep and Advanced models are built in one of two ways: where the model works on the fields directly its explanations read the same as Core's, and where it works on a transformed version of the data they do not. An Advanced model of the second kind names the fields its learned representation leans on, reported as leads rather than as attribution, so those rows carry signals alongside an interpretable_share at or near 0. A Deep model of the second kind works in derived dimensions with no route back to individual fields, so instead of a breakdown it reports a measurement: each field the host presented is removed, the prediction is re-run on the same model, and the weight is how far the result moved. Those figures do not sum to the score, so they are excluded from interpretable_share, which reads 0 on such a row while the signals themselves are fully named. Fields the model reads nothing from are listed under observations_without_effect rather than ranked, and a row whose fields each independently place the host carries a redundant_evidence caveat — no single removal moves it, which is agreement rather than weak evidence. Read the row rather than the tier: ranking_basis tells you which case you got.
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", [])
if kind == "out_of_distribution":
print(f"IP {ip}: out-of-distribution (no fingerprint match)")
elif kind == "no_input_features":
print(f"IP {ip}: no readable input - check the raw scan data")
elif kind == "confident_match":
print(f"IP {ip}: Cluster {cluster} (confident, confidence={confidence:.2f})")
else:
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.
