Direct Python Import

All OSINT AI One tools are independent async functions. You can import them directly into your own code without needing CLI, server or agent.

Install as dependency

pip install -e /path/to/osint-ai-one
# or if published on PyPI:
pip install osint-ai-one

Individual tools

import asyncio
from src.tools.virustotal import virustotal_ip_lookup, virustotal_domain_lookup
from src.tools.abuseipdb import abuseipdb_check_ip
from src.tools.shodan_tool import shodan_host_lookup
from src.tools.ipinfo import ipinfo_lookup
from src.tools.dns_tool import dns_lookup
from src.tools.whois_tool import whois_lookup
from src.tools.crtsh import crtsh_lookup
from src.tools.gdelt_tool import gdelt_entity_search
from src.tools.country_risk import get_country_risk_score

async def main():
    # IP analysis
    vt_result = await virustotal_ip_lookup("185.220.101.34")
    abuse_result = await abuseipdb_check_ip("185.220.101.34")
    shodan_result = await shodan_host_lookup("185.220.101.34")

    # Domain analysis
    dns_result = await dns_lookup("acme-corp.com")
    whois_result = await whois_lookup("acme-corp.com")
    subdomains = await crtsh_lookup("acme-corp.com")

    # Context
    news = await gdelt_entity_search("Acme Corp")
    risk = await get_country_risk_score("ES")

asyncio.run(main())

Parallel analysis

Use asyncio.gather for parallel calls and maximize speed:

import asyncio
from src.tools.virustotal import virustotal_ip_lookup
from src.tools.abuseipdb import abuseipdb_check_ip
from src.tools.shodan_tool import shodan_host_lookup
from src.tools.alienvault import alienvault_ip_lookup
from src.tools.ipinfo import ipinfo_lookup

async def analyze_ip(ip: str) -> dict:
    results = await asyncio.gather(
        virustotal_ip_lookup(ip),
        abuseipdb_check_ip(ip),
        shodan_host_lookup(ip),
        alienvault_ip_lookup(ip),
        ipinfo_lookup(ip),
        return_exceptions=True
    )
    return {
        "virustotal": results[0],
        "abuseipdb": results[1],
        "shodan": results[2],
        "alienvault": results[3],
        "ipinfo": results[4],
    }

result = asyncio.run(analyze_ip("185.220.101.34"))

Use unified facade

For access to all functionality via a single class:

from src.services.osint_service import OsintService

service = OsintService()

# Investigate an IOC with the complete agent
result = await service.investigate("185.220.101.34")
print(result.assessment)
print(result.risk_score)

# Risk score only
score = await service.get_risk_score("185.220.101.34")

# Create and manage investigations
investigation = await service.create_investigation(
    name="My investigation",
    goal="Due diligence"
)

Embed in a FastAPI application

from fastapi import FastAPI
from src.tools.virustotal import virustotal_ip_lookup
from src.scoring.dashboard import calculate_risk_score

app = FastAPI()

@app.get("/analyze/ip/{ip}")
async def analyze_ip(ip: str):
    vt_result = await virustotal_ip_lookup(ip)
    score = calculate_risk_score({"virustotal": vt_result})
    return {
        "ip": ip,
        "risk_score": score,
        "malicious": score > 60,
        "details": vt_result
    }

Environment variables

Tools automatically read API keys from environment variables. Load your .env before importing:

from dotenv import load_dotenv
load_dotenv("/path/to/osint-ai-one/.env")

from src.tools.virustotal import virustotal_ip_lookup

Custom batch processing

import asyncio
from src.tools.virustotal import virustotal_ip_lookup
from src.cache import cached  # Cache decorator

async def process_ioc_list(iocs: list[str], delay: float = 2.0) -> list[dict]:
    results = []
    for ioc in iocs:
        try:
            result = await virustotal_ip_lookup(ioc)
            results.append({"ioc": ioc, "result": result, "error": None})
        except Exception as e:
            results.append({"ioc": ioc, "result": None, "error": str(e)})
        await asyncio.sleep(delay)  # Respect rate limits
    return results

iocs = ["185.220.101.34", "8.8.8.8", "evil.com"]
results = asyncio.run(process_ioc_list(iocs))