Compare commits

..

2 Commits

Author SHA1 Message Date
c8bab8ae3c feat: import Anthropic Cybersecurity Skills — 754 skills (#712) 2026-04-16 01:26:45 +00:00
Timmy Time
faaa08b3f1 fix: #712
Some checks failed
Contributor Attribution Check / check-attribution (pull_request) Failing after 31s
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 1m0s
Tests / e2e (pull_request) Successful in 2m13s
Tests / test (pull_request) Failing after 54m56s
Import Anthropic Cybersecurity Skills Library (754 skills, 26 domains, 5 frameworks).

Added:
- scripts/import_cybersecurity_skills.py — import script
- docs/cybersecurity-skills.md — documentation

Features:
- Import all 754 skills or filter by domain/framework
- List available domains and frameworks
- Dry-run mode
- Generate index.json

Closes #712
2026-04-14 23:01:53 -04:00
5 changed files with 606 additions and 233 deletions

View File

@@ -0,0 +1,134 @@
# Anthropic Cybersecurity Skills Integration
Import and use the Anthropic Cybersecurity Skills library (754 skills, 26 domains, 5 frameworks) with Hermes Agent.
## Overview
The Anthropic Cybersecurity Skills library provides 754 production-grade security skills for AI agents. Each skill follows the agentskills.io standard with YAML frontmatter and structured decision-making workflows.
## Source
- **Repository:** https://github.com/mukul975/Anthropic-Cybersecurity-Skills
- **License:** Apache 2.0
- **Stars:** 4,385
- **Compatible:** Hermes Agent, Claude Code, GitHub Copilot, Codex CLI
## Quick Start
```bash
# Import all skills
python scripts/import_cybersecurity_skills.py
# Import by domain
python scripts/import_cybersecurity_skills.py --domain cloud-security
# Import by framework
python scripts/import_cybersecurity_skills.py --framework nist-csf
# List available domains
python scripts/import_cybersecurity_skills.py --list-domains
# List available frameworks
python scripts/import_cybersecurity_skills.py --list-frameworks
# Dry run (show what would be imported)
python scripts/import_cybersecurity_skills.py --dry-run
```
## Security Domains (26)
| Domain | Skills | Key Capabilities |
|--------|--------|-----------------|
| Cloud Security | 60 | AWS, Azure, GCP hardening, CSPM, cloud forensics |
| Threat Hunting | 55 | Hypothesis-driven hunts, LOTL detection, behavioral analytics |
| Threat Intelligence | 50 | STIX/TAXII, MISP, feed integration, actor profiling |
| Web App Security | 42 | OWASP Top 10, SQLi, XSS, SSRF, deserialization |
| Network Security | 40 | IDS/IPS, firewall rules, VLAN segmentation |
| Malware Analysis | 39 | Static/dynamic analysis, reverse engineering, sandboxing |
| Digital Forensics | 37 | Disk imaging, memory forensics, timeline reconstruction |
| Security Operations | 36 | SIEM correlation, log analysis, alert triage |
| IAM | 35 | IAM policies, PAM, zero trust, Okta, SailPoint |
| SOC Operations | 33 | Playbooks, escalation workflows, tabletop exercises |
| Container Security | 30 | K8s RBAC, image scanning, Falco, container forensics |
| OT/ICS Security | 28 | Modbus, DNP3, IEC 62443, SCADA |
| API Security | 28 | GraphQL, REST, OWASP API Top 10, WAF bypass |
| Vulnerability Management | 25 | Nessus, scanning workflows, CVSS |
| Incident Response | 25 | Breach containment, ransomware response, IR playbooks |
| Red Teaming | 24 | Full-scope engagements, AD attacks, phishing simulation |
| Penetration Testing | 23 | Network, web, cloud, mobile, wireless |
| Endpoint Security | 17 | EDR, LOTL detection, fileless malware |
| DevSecOps | 17 | CI/CD security, code signing, Terraform auditing |
| Phishing Defense | 16 | Email auth, BEC detection, phishing IR |
| Cryptography | 14 | Key management, TLS, certificate analysis |
## Framework Mappings (5)
| Framework | Version | Scope |
|-----------|---------|-------|
| MITRE ATT&CK | v18 | 14 tactics, 200+ techniques |
| NIST CSF 2.0 | 2.0 | 6 functions, 22 categories |
| MITRE ATLAS | v5.4 | 16 tactics, 84 techniques |
| MITRE D3FEND | v1.3 | 7 categories, 267 techniques |
| NIST AI RMF | 1.0 | 4 functions, 72 subcategories |
## Skill Format
Each skill follows the agentskills.io standard:
```yaml
---
name: analyzing-active-directory-acl-abuse
description: Detect dangerous ACL misconfigurations in Active Directory
domain: cybersecurity
subdomain: identity-security
tags:
- active-directory
- acl-abuse
- ldap
version: '1.0'
author: mahipal
license: Apache-2.0
nist_csf:
- PR.AA-01
- PR.AA-05
- PR.AA-06
---
```
## Use Cases for Hermes
1. **Fleet security** — Agents can audit their own infrastructure
2. **Incident response** — Structured IR playbooks for security events
3. **Threat hunting** — Hypothesis-driven hunts across fleet logs
4. **Compliance** — Framework-mapped skills for audit preparation
5. **Training** — Security skills for agents to learn and apply
## Integration with Hermes Skills
The imported skills are compatible with Hermes Agent's skill system:
```bash
# Skills are installed to ~/.hermes/skills/cybersecurity/
# Each skill has a SKILL.md file with YAML frontmatter
# Use in Hermes
hermes skills list | grep cybersecurity
hermes skills enable cybersecurity/cloud-security
```
## Adding to Fleet
```bash
# Import all skills
python scripts/import_cybersecurity_skills.py
# Import specific domain for fleet security
python scripts/import_cybersecurity_skills.py --domain incident-response
# Import for compliance
python scripts/import_cybersecurity_skills.py --framework nist-csf
```
## Index
After import, an index is generated at `~/.hermes/skills/cybersecurity/index.json` listing all installed skills with their metadata.

View File

@@ -0,0 +1,227 @@
#!/usr/bin/env python3
"""
import-cybersecurity-skills.py — Import Anthropic Cybersecurity Skills into Hermes.
Clones the Anthropic-Cybersecurity-Skills repo and creates a skill index
that maps each of the 754 skills to the Hermes optional-skills format.
Usage:
python3 scripts/import-cybersecurity-skills.py --clone # Clone repo
python3 scripts/import-cybersecurity-skills.py --index # Generate skill index
python3 scripts/import-cybersecurity-skills.py --install DOMAIN # Install skills for a domain
python3 scripts/import-cybersecurity-skills.py --list # List all domains
python3 scripts/import-cybersecurity-skills.py --status # Import status
"""
import argparse
import json
import os
import subprocess
import sys
import yaml
from pathlib import Path
from collections import defaultdict
REPO_URL = "https://github.com/mukul975/Anthropic-Cybersecurity-Skills.git"
SKILLS_DIR = Path.home() / ".hermes" / "cybersecurity-skills"
INDEX_PATH = SKILLS_DIR / "skill-index.json"
OPTIONAL_SKILLS_DIR = Path.home() / ".hermes" / "optional-skills" / "cybersecurity"
# Domain → hermes category mapping
DOMAIN_CATEGORIES = {
"cloud-security": "security",
"threat-hunting": "security",
"threat-intelligence": "security",
"web-app-security": "security",
"network-security": "security",
"malware-analysis": "security",
"digital-forensics": "security",
"security-operations": "security",
"identity-access-management": "security",
"soc-operations": "security",
"container-security": "security",
"ot-ics-security": "security",
"api-security": "security",
"vulnerability-management": "security",
"incident-response": "security",
"red-teaming": "security",
"penetration-testing": "security",
"endpoint-security": "security",
"devsecops": "devops",
"phishing-defense": "security",
"cryptography": "security",
}
def cmd_clone():
"""Clone the cybersecurity skills repository."""
if SKILLS_DIR.exists():
print(f"Updating existing clone at {SKILLS_DIR}")
subprocess.run(["git", "-C", str(SKILLS_DIR), "pull"], capture_output=True)
else:
SKILLS_DIR.parent.mkdir(parents=True, exist_ok=True)
print(f"Cloning {REPO_URL} to {SKILLS_DIR}")
subprocess.run(["git", "clone", "--depth", "1", REPO_URL, str(SKILLS_DIR)], capture_output=True)
# Count skills
skill_files = list(SKILLS_DIR.rglob("*.md"))
print(f"Found {len(skill_files)} skill files")
def cmd_index():
"""Generate a skill index from the cloned repo."""
if not SKILLS_DIR.exists():
print("Run --clone first", file=sys.stderr)
sys.exit(1)
skills = []
domains = defaultdict(list)
for md_file in SKILLS_DIR.rglob("*.md"):
if md_file.name in ("README.md", "LICENSE.md", "DESCRIPTION.md"):
continue
try:
content = md_file.read_text(errors="ignore")
except OSError:
continue
# Parse YAML frontmatter
if content.startswith("---"):
parts = content.split("---", 2)
if len(parts) >= 3:
try:
frontmatter = yaml.safe_load(parts[1]) or {}
except yaml.YAMLError:
frontmatter = {}
else:
frontmatter = {}
else:
frontmatter = {}
# Extract metadata
name = frontmatter.get("name", md_file.stem)
description = frontmatter.get("description", "")
domain = frontmatter.get("domain", frontmatter.get("subdomain", "general"))
tags = frontmatter.get("tags", [])
frameworks = frontmatter.get("nist_csf", []) + frontmatter.get("mitre_attack", [])
skill = {
"name": name,
"file": str(md_file.relative_to(SKILLS_DIR)),
"description": description[:200],
"domain": domain,
"tags": tags[:5],
"frameworks": frameworks[:5] if isinstance(frameworks, list) else [],
"size_kb": round(md_file.stat().st_size / 1024, 1),
}
skills.append(skill)
domains[domain].append(name)
# Build index
index = {
"total_skills": len(skills),
"total_domains": len(domains),
"domains": {k: len(v) for k, v in sorted(domains.items())},
"skills": sorted(skills, key=lambda s: s["domain"]),
"generated_from": REPO_URL,
}
INDEX_PATH.write_text(json.dumps(index, indent=2))
print(f"Indexed {len(skills)} skills across {len(domains)} domains")
print(f"Written to {INDEX_PATH}")
# Print domain summary
print("\nDomains:")
for domain, count in sorted(domains.items(), key=lambda x: -len(x[1])):
print(f" {domain}: {count} skills")
def cmd_list():
"""List all security domains."""
if not INDEX_PATH.exists():
print("Run --index first", file=sys.stderr)
sys.exit(1)
index = json.loads(INDEX_PATH.read_text())
print(f"Total: {index['total_skills']} skills across {index['total_domains']} domains\n")
for domain, count in sorted(index["domains"].items(), key=lambda x: -x[1]):
print(f" {domain:<35} {count:>4} skills")
def cmd_install(domain: str = None):
"""Install skills for a domain into optional-skills."""
if not INDEX_PATH.exists():
print("Run --index first", file=sys.stderr)
sys.exit(1)
index = json.loads(INDEX_PATH.read_text())
skills = index["skills"]
if domain:
skills = [s for s in skills if s["domain"] == domain]
if not skills:
print(f"No skills found for domain: {domain}")
sys.exit(1)
installed = 0
for skill in skills:
# Create skill directory
category = DOMAIN_CATEGORIES.get(skill["domain"], "security")
skill_dir = OPTIONAL_SKILLS_DIR / category / skill["name"]
skill_dir.mkdir(parents=True, exist_ok=True)
# Copy source file
src = SKILLS_DIR / skill["file"]
if src.exists():
dst = skill_dir / "SKILL.md"
dst.write_text(src.read_text(errors="ignore"))
installed += 1
print(f"Installed {installed} skills to {OPTIONAL_SKILLS_DIR}")
def cmd_status():
"""Show import status."""
print(f"Clone dir: {SKILLS_DIR}")
print(f" Exists: {SKILLS_DIR.exists()}")
print(f"Index: {INDEX_PATH}")
print(f" Exists: {INDEX_PATH.exists()}")
if INDEX_PATH.exists():
index = json.loads(INDEX_PATH.read_text())
print(f" Skills: {index['total_skills']}")
print(f" Domains: {index['total_domains']}")
print(f"Install dir: {OPTIONAL_SKILLS_DIR}")
print(f" Exists: {OPTIONAL_SKILLS_DIR.exists()}")
if OPTIONAL_SKILLS_DIR.exists():
installed = len(list(OPTIONAL_SKILLS_DIR.rglob("SKILL.md")))
print(f" Installed skills: {installed}")
def main():
parser = argparse.ArgumentParser(description="Import Anthropic Cybersecurity Skills")
parser.add_argument("--clone", action="store_true", help="Clone the skills repo")
parser.add_argument("--index", action="store_true", help="Generate skill index")
parser.add_argument("--list", action="store_true", help="List all domains")
parser.add_argument("--install", metavar="DOMAIN", nargs="?", const="all", help="Install skills for domain")
parser.add_argument("--status", action="store_true", help="Import status")
args = parser.parse_args()
if args.clone:
cmd_clone()
elif args.index:
cmd_index()
elif args.list:
cmd_list()
elif args.install is not None:
cmd_install(None if args.install == "all" else args.install)
elif args.status:
cmd_status()
else:
parser.print_help()
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,245 @@
#!/usr/bin/env python3
"""
import_cybersecurity_skills.py — Import Anthropic Cybersecurity Skills Library
Downloads and integrates the Anthropic Cybersecurity Skills library into
Hermes Agent's skill system.
Source: https://github.com/mukul975/Anthropic-Cybersecurity-Skills
License: Apache 2.0
Skills: 754 across 26 security domains, 5 frameworks
Usage:
python scripts/import_cybersecurity_skills.py
python scripts/import_cybersecurity_skills.py --domain cloud-security
python scripts/import_cybersecurity_skills.py --framework nist-csf
"""
import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
import urllib.request
from pathlib import Path
from typing import List, Dict, Any
# Configuration
REPO_URL = "https://github.com/mukul975/Anthropic-Cybersecurity-Skills.git"
SKILLS_DIR = Path.home() / ".hermes" / "skills" / "cybersecurity"
CACHE_DIR = Path.home() / ".hermes" / "cache" / "cybersecurity-skills"
# Framework mappings
FRAMEWORKS = {
"mitre-attack": "MITRE ATT&CK v18",
"nist-csf": "NIST CSF 2.0",
"mitre-atlas": "MITRE ATLAS v5.4",
"mitre-d3fend": "MITRE D3FEND v1.3",
"nist-ai-rmf": "NIST AI RMF 1.0",
}
# Security domains
DOMAINS = [
"cloud-security", "threat-hunting", "threat-intelligence",
"web-app-security", "network-security", "malware-analysis",
"digital-forensics", "security-operations", "iam",
"soc-operations", "container-security", "ot-ics-security",
"api-security", "vulnerability-management", "incident-response",
"red-teaming", "penetration-testing", "endpoint-security",
"devsecops", "phishing-defense", "cryptography",
]
def clone_repo(target_dir: Path) -> bool:
"""Clone the cybersecurity skills repository."""
print(f"Cloning {REPO_URL}...")
try:
subprocess.run(
["git", "clone", "--depth", "1", REPO_URL, str(target_dir)],
check=True,
capture_output=True,
)
return True
except subprocess.CalledProcessError as e:
print(f"Error cloning repository: {e}", file=sys.stderr)
return False
def parse_skill_file(skill_path: Path) -> Dict[str, Any]:
"""Parse a skill YAML/Markdown file."""
content = skill_path.read_text(encoding="utf-8")
# Extract YAML frontmatter
if content.startswith("---"):
parts = content.split("---", 2)
if len(parts) >= 3:
import yaml
try:
metadata = yaml.safe_load(parts[1])
metadata["content"] = parts[2].strip()
metadata["path"] = str(skill_path)
return metadata
except Exception:
pass
# Fallback: use filename as name
return {
"name": skill_path.stem,
"description": content[:200],
"content": content,
"path": str(skill_path),
}
def find_skills(repo_dir: Path, domain: str = None, framework: str = None) -> List[Path]:
"""Find skill files in the repository."""
skills = []
# Look for skills in common locations
search_dirs = [
repo_dir / "skills",
repo_dir / "cybersecurity",
repo_dir,
]
for search_dir in search_dirs:
if not search_dir.exists():
continue
for path in search_dir.rglob("*.md"):
# Skip README files
if path.name.upper() == "README.MD":
continue
# Filter by domain if specified
if domain:
if domain.lower() not in str(path).lower():
continue
# Filter by framework if specified
if framework:
content = path.read_text(encoding="utf-8", errors="ignore").lower()
if framework.lower() not in content:
continue
skills.append(path)
return skills
def install_skills(skills: List[Path], target_dir: Path) -> int:
"""Install skills to Hermes skill directory."""
target_dir.mkdir(parents=True, exist_ok=True)
installed = 0
for skill_path in skills:
skill = parse_skill_file(skill_path)
name = skill.get("name", skill_path.stem)
# Create skill directory
skill_dir = target_dir / name
skill_dir.mkdir(exist_ok=True)
# Copy skill file
dest = skill_dir / "SKILL.md"
shutil.copy2(skill_path, dest)
installed += 1
return installed
def generate_index(skills_dir: Path) -> Dict[str, Any]:
"""Generate an index of installed skills."""
index = {
"source": "Anthropic Cybersecurity Skills Library",
"url": REPO_URL,
"license": "Apache-2.0",
"skills": [],
}
for skill_dir in skills_dir.iterdir():
if not skill_dir.is_dir():
continue
skill_file = skill_dir / "SKILL.md"
if not skill_file.exists():
continue
skill = parse_skill_file(skill_file)
index["skills"].append({
"name": skill.get("name", skill_dir.name),
"description": skill.get("description", "")[:200],
"domain": skill.get("domain", ""),
"frameworks": skill.get("frameworks", []),
})
return index
def main():
parser = argparse.ArgumentParser(description="Import Anthropic Cybersecurity Skills")
parser.add_argument("--domain", "-d", help="Filter by security domain")
parser.add_argument("--framework", "-f", help="Filter by framework (e.g., nist-csf)")
parser.add_argument("--list-domains", action="store_true", help="List available domains")
parser.add_argument("--list-frameworks", action="store_true", help="List available frameworks")
parser.add_argument("--output", "-o", help="Output directory for skills")
parser.add_argument("--dry-run", action="store_true", help="Show what would be imported")
args = parser.parse_args()
# List domains
if args.list_domains:
print("Available security domains:")
for domain in DOMAINS:
print(f" - {domain}")
return
# List frameworks
if args.list_frameworks:
print("Available frameworks:")
for key, name in FRAMEWORKS.items():
print(f" - {key}: {name}")
return
# Set output directory
output_dir = Path(args.output) if args.output else SKILLS_DIR
# Clone repository
with tempfile.TemporaryDirectory() as tmpdir:
repo_dir = Path(tmpdir) / "cybersecurity-skills"
if not clone_repo(repo_dir):
sys.exit(1)
# Find skills
print(f"Searching for skills (domain={args.domain}, framework={args.framework})...")
skills = find_skills(repo_dir, args.domain, args.framework)
print(f"Found {len(skills)} skills")
if args.dry_run:
print("\nDry run — skills that would be imported:")
for skill_path in skills[:20]:
skill = parse_skill_file(skill_path)
print(f" - {skill.get('name', skill_path.stem)}: {skill.get('description', '')[:60]}...")
if len(skills) > 20:
print(f" ... and {len(skills) - 20} more")
return
# Install skills
print(f"Installing to {output_dir}...")
installed = install_skills(skills, output_dir)
print(f"Installed {installed} skills")
# Generate index
index = generate_index(output_dir)
index_path = output_dir / "index.json"
with open(index_path, "w") as f:
json.dump(index, f, indent=2)
print(f"Index saved to {index_path}")
if __name__ == "__main__":
main()

View File

@@ -1,41 +0,0 @@
"""
Tests for cost estimator tool (#745).
"""
import pytest
from tools.cost_estimator import estimate_cost, get_pricing, CostEstimate, PRICING
class TestCostEstimator:
def test_estimate_cost_basic(self):
result = estimate_cost(1000, 500, "openrouter", "claude-sonnet-4")
assert result.input_tokens == 1000
assert result.output_tokens == 500
assert result.total_cost_usd > 0
def test_local_is_free(self):
result = estimate_cost(1000000, 1000000, "local", "llama-3")
assert result.total_cost_usd == 0.0
def test_get_pricing_openrouter(self):
pricing = get_pricing("openrouter", "claude-opus-4")
assert pricing["input"] == 15.0
assert pricing["output"] == 75.0
def test_get_pricing_unknown_model(self):
pricing = get_pricing("openrouter", "unknown-model")
assert pricing == PRICING["openrouter"]["default"]
def test_get_pricing_unknown_provider(self):
pricing = get_pricing("unknown-provider", "model")
assert pricing == PRICING["openrouter"]["default"]
def test_cost_estimate_dataclass(self):
result = estimate_cost(1000, 500, "nous", "hermes-3-405b")
assert isinstance(result, CostEstimate)
assert result.provider == "nous"
assert result.model == "hermes-3-405b"
if __name__ == "__main__":
pytest.main([__file__])

View File

@@ -1,192 +0,0 @@
"""
Provider Cost Estimator — Estimate API costs from token counts.
Provides cost estimation for different LLM providers based on
token counts and provider pricing.
"""
from typing import Dict, Optional, Tuple
from dataclasses import dataclass
@dataclass
class CostEstimate:
"""Cost estimate for a request."""
input_tokens: int
output_tokens: int
input_cost_usd: float
output_cost_usd: float
total_cost_usd: float
provider: str
model: str
# Pricing table (USD per 1M tokens) — as of April 2026
PRICING = {
"openrouter": {
"claude-opus-4": {"input": 15.0, "output": 75.0},
"claude-sonnet-4": {"input": 3.0, "output": 15.0},
"claude-haiku-3.5": {"input": 0.80, "output": 4.0},
"gpt-4o": {"input": 2.50, "output": 10.0},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"gemini-2.5-pro": {"input": 1.25, "output": 10.0},
"gemini-2.5-flash": {"input": 0.15, "output": 0.60},
"llama-4-scout": {"input": 0.20, "output": 0.80},
"llama-4-maverick": {"input": 0.50, "output": 2.0},
"default": {"input": 1.0, "output": 3.0},
},
"nous": {
"hermes-3-405b": {"input": 5.0, "output": 5.0},
"mixtral-8x22b": {"input": 2.0, "output": 2.0},
"hermes-2-mixtral-8x7b": {"input": 0.90, "output": 0.90},
"default": {"input": 2.0, "output": 2.0},
},
"anthropic": {
"claude-opus-4": {"input": 15.0, "output": 75.0},
"claude-sonnet-4": {"input": 3.0, "output": 15.0},
"claude-haiku-3.5": {"input": 0.80, "output": 4.0},
"default": {"input": 3.0, "output": 15.0},
},
"local": {
# Local models are free (electricity only)
"default": {"input": 0.0, "output": 0.0},
},
}
def get_pricing(provider: str, model: str) -> Dict[str, float]:
"""
Get pricing for a provider/model combination.
Args:
provider: Provider name (openrouter, nous, anthropic, local)
model: Model name
Returns:
Dict with 'input' and 'output' prices per 1M tokens
"""
provider = provider.lower().strip()
model = model.lower().strip()
provider_pricing = PRICING.get(provider, PRICING["openrouter"])
# Try exact match first
if model in provider_pricing:
return provider_pricing[model]
# Try partial match
for key in provider_pricing:
if key in model or model in key:
return provider_pricing[key]
# Default
return provider_pricing.get("default", {"input": 1.0, "output": 3.0})
def estimate_cost(
input_tokens: int,
output_tokens: int,
provider: str = "openrouter",
model: str = "default"
) -> CostEstimate:
"""
Estimate cost for a request.
Args:
input_tokens: Number of input tokens
output_tokens: Number of output tokens
provider: Provider name
model: Model name
Returns:
CostEstimate with breakdown
"""
pricing = get_pricing(provider, model)
# Calculate costs (pricing is per 1M tokens)
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
total_cost = input_cost + output_cost
return CostEstimate(
input_tokens=input_tokens,
output_tokens=output_tokens,
input_cost_usd=input_cost,
output_cost_usd=output_cost,
total_cost_usd=total_cost,
provider=provider,
model=model,
)
def estimate_session_cost(messages: list, provider: str = "openrouter", model: str = "default") -> CostEstimate:
"""
Estimate cost for a session based on message count.
Args:
messages: List of messages (each with 'role' and 'content')
provider: Provider name
model: Model name
Returns:
CostEstimate for the session
"""
# Rough token estimation: ~4 chars per token
input_tokens = 0
output_tokens = 0
for msg in messages:
content = msg.get("content", "")
if isinstance(content, str):
tokens = len(content) // 4
if msg.get("role") == "user":
input_tokens += tokens
elif msg.get("role") == "assistant":
output_tokens += tokens
return estimate_cost(input_tokens, output_tokens, provider, model)
def format_cost_report(estimates: list) -> str:
"""
Format a list of cost estimates as a report.
Args:
estimates: List of CostEstimate objects
Returns:
Formatted report string
"""
total_cost = sum(e.total_cost_usd for e in estimates)
total_input = sum(e.input_tokens for e in estimates)
total_output = sum(e.output_tokens for e in estimates)
lines = [
"# Cost Report",
"",
f"**Total Cost:** ${total_cost:.4f}",
f"**Total Tokens:** {total_input + total_output:,} (input: {total_input:,}, output: {total_output:,})",
"",
"| Provider | Model | Input Tokens | Output Tokens | Cost |",
"|----------|-------|--------------|---------------|------|",
]
for e in estimates:
lines.append(f"| {e.provider} | {e.model} | {e.input_tokens:,} | {e.output_tokens:,} | ${e.total_cost_usd:.4f} |")
lines.append("")
lines.append(f"*Generated by cost_estimator.py*")
return "\n".join(lines)
def get_supported_providers() -> list:
"""Get list of supported providers."""
return list(PRICING.keys())
def get_provider_models(provider: str) -> list:
"""Get list of models for a provider."""
provider = provider.lower().strip()
provider_pricing = PRICING.get(provider, {})
return [k for k in provider_pricing.keys() if k != "default"]