Multi-Agent with A2A

The multi-agent pattern

In a multi-agent system, an orchestrator agent divides work and delegates subtasks to specialized agents. OSINT AI One acts as a specialized threat intelligence agent.

Orchestrator Agent
    ├─ "Analyze IP 185.220.101.34" → OSINT AI One (A2A)
    ├─ "Search JIRA for related incidents" → JIRA Agent
    └─ "Notify Slack if critical" → Slack Agent

Start A2A server

osint-a2a               # 0.0.0.0:9000
osint-a2a --port 9090   # custom port

Call from another Python agent

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

With event streaming

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']}] Querying...")
        elif event.type == "tool_result":
            print(f"[{event.data['tool']}] Result received")
        elif event.type == "final":
            print(f"Risk: {event.data['risk_score']}/100")
            return event.data

The 5 available skills

Skill	Example input	Returns
investigate_ip	"Investigate IP 185.220.101.34"	Complete analysis + risk score
investigate_domain	"investigate domain evil.com"	Complete analysis + risk score
investigate_url	"analyze URL http://..."	URL analysis + verdict
threat_feed_analysis	"analyze feodo feed top 5"	List of IOCs with evaluation
risk_scoring	"score 185.220.101.34"	Only score 0-100

Use case: automated triage pipeline

import asyncio
from a2a_sdk import A2AClient

async def triage_alerts(alerts: list[dict]) -> list[dict]:
    """Enriches SIEM alerts with OSINT intelligence."""
    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)

# Usage
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))

Use case: orchestrator agent with Claude

If you have a Claude-based agent with MCP or A2A:

# The Claude agent can delegate to OSINT AI One automatically
# when it detects a need for threat intelligence:

# In the orchestrator agent's system prompt:
"""
When you need to investigate an IP, domain or URL, use the
OSINT AI Agent available at http://localhost:9000 via A2A.
Available skills: investigate_ip, investigate_domain,
investigate_url, threat_feed_analysis, risk_scoring.
"""

Agent Card

Any A2A-compatible agent can auto-discover OSINT AI One capabilities:

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"]
}