Multi-agente con A2A

El patron multi-agente

En un sistema multi-agente, un agente orquestador divide el trabajo y delega subtareas a agentes especializados. OSINT AI One actua como agente especializado en threat intelligence.

Agente Orquestador
    ├─ "Analiza la IP 185.220.101.34" → OSINT AI One (A2A)
    ├─ "Busca en JIRA incidentes relacionados" → Agente JIRA
    └─ "Notifica a Slack si es critico" → Agente Slack

Arrancar el servidor A2A

osint-a2a               # 0.0.0.0:9000
osint-a2a --port 9090   # puerto personalizado

Llamada desde otro agente Python

from a2a_sdk import A2AClient

async def investigate_with_osint(ip: str) -> dict:
    client = A2AClient("http://localhost:9000")

    result = await client.send_task({
        "skill": "investigate_ip",
        "message": f"Investigate IP {ip}"
    })

    return {
        "ip": ip,
        "assessment": result.get("assessment"),
        "risk_score": result.get("risk_score"),
        "risk_level": result.get("risk_level")
    }

Con streaming de eventos

async def investigate_with_progress(ip: str):
    client = A2AClient("http://localhost:9000")

    async for event in client.stream_task({
        "skill": "investigate_ip",
        "message": f"Investigate IP {ip}"
    }):
        if event.type == "tool_call":
            print(f"[{event.data['tool']}] Consultando...")
        elif event.type == "tool_result":
            print(f"[{event.data['tool']}] Resultado recibido")
        elif event.type == "final":
            print(f"Riesgo: {event.data['risk_score']}/100")
            return event.data

Las 5 skills disponibles

Skill	Input ejemplo	Devuelve
investigate_ip	"Investigate IP 185.220.101.34"	Analisis completo + risk score
investigate_domain	"investigate domain evil.com"	Analisis completo + risk score
investigate_url	"analyze URL http://..."	Analisis URL + veredicto
threat_feed_analysis	"analyze feodo feed top 5"	Lista de IOCs con evaluacion
risk_scoring	"score 185.220.101.34"	Solo el score 0-100

Caso de uso: pipeline de triaje automatico

import asyncio
from a2a_sdk import A2AClient

async def triage_alerts(alerts: list[dict]) -> list[dict]:
    """Enriquece alertas del SIEM con inteligencia OSINT."""
    client = A2AClient("http://localhost:9000")
    results = []

    for alert in alerts:
        ioc = alert.get("src_ip") or alert.get("domain")
        if not ioc:
            continue

        skill = "investigate_ip" if "." in ioc and not any(c.isalpha() for c in ioc.replace(".", "")) \
                else "investigate_domain"

        result = await client.send_task({
            "skill": skill,
            "message": f"Quick risk assessment for {ioc}"
        })

        results.append({
            **alert,
            "osint_risk_score": result.get("risk_score", 0),
            "osint_risk_level": result.get("risk_level", "UNKNOWN"),
            "should_escalate": result.get("risk_score", 0) >= 70
        })

        await asyncio.sleep(1)  # Rate limiting

    return sorted(results, key=lambda x: x["osint_risk_score"], reverse=True)

# Uso
alerts = [
    {"alert_id": "ALT-001", "src_ip": "185.220.101.34"},
    {"alert_id": "ALT-002", "domain": "phishing-example.com"},
    {"alert_id": "ALT-003", "src_ip": "8.8.8.8"},
]
enriched = asyncio.run(triage_alerts(alerts))

Caso de uso: agente orquestador con Claude

Si tienes un agente basado en Claude con MCP o A2A:

# El agente Claude puede delegar a OSINT AI One automaticamente
# cuando detecta que necesita inteligencia de amenazas:

# En el system prompt del agente orquestador:
"""
Cuando necesites investigar una IP, dominio o URL, usa el
OSINT AI Agent disponible en http://localhost:9000 via A2A.
Skills disponibles: investigate_ip, investigate_domain,
investigate_url, threat_feed_analysis, risk_scoring.
"""

Agent Card

Cualquier agente compatible con A2A puede autodescubrir las capacidades de OSINT AI One:

curl http://localhost:9000/.well-known/agent-card.json

{
  "name": "OSINT AI Agent",
  "description": "Autonomous threat intelligence platform with 28 OSINT tools",
  "skills": ["investigate_ip", "investigate_domain", "investigate_url",
             "threat_feed_analysis", "risk_scoring"],
  "streaming": true,
  "output_modes": ["json", "text"]
}