Python import directo

Todas las herramientas de OSINT AI One son funciones async independientes. Puedes importarlas directamente en tu propio codigo sin necesidad de CLI, servidor ni agente.

Instalacion como dependencia

pip install -e /ruta/a/osint-ai-one
# o si esta publicado en PyPI:
pip install osint-ai-one

Herramientas individuales

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():
    # Analisis de IP
    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")

    # Analisis de dominio
    dns_result = await dns_lookup("acme-corp.com")
    whois_result = await whois_lookup("acme-corp.com")
    subdomains = await crtsh_lookup("acme-corp.com")

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

asyncio.run(main())

Analisis paralelo

Usa asyncio.gather para llamadas en paralelo y maximizar velocidad:

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"))

Usar la fachada unificada

Para acceso a toda la funcionalidad via una sola clase:

from src.services.osint_service import OsintService

service = OsintService()

# Investigar un IOC con el agente completo
result = await service.investigate("185.220.101.34")
print(result.assessment)
print(result.risk_score)

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

# Crear y gestionar investigaciones
investigation = await service.create_investigation(
    name="Mi investigacion",
    goal="Due diligence"
)

Incrustar en una aplicacion FastAPI

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
    }

Variables de entorno

Las herramientas leen las API keys de variables de entorno automaticamente. Carga tu .env antes de importar:

from dotenv import load_dotenv
load_dotenv("/ruta/a/osint-ai-one/.env")

from src.tools.virustotal import virustotal_ip_lookup

Procesamiento batch personalizado

import asyncio
from src.tools.virustotal import virustotal_ip_lookup
from src.cache import cached  # Decorador de cache

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)  # Respetar rate limits
    return results

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