Partner Integration Guide

Connect your AI agent or LLM to DEFONEOS via the MCP (Model Context Protocol) federation. This is the native integration — your agent discovers and calls DEFONEOS tools the same way it calls any MCP server.

Python — Connect to DEFONEOS MCP Federation

# Install the SDK
pip install defoneos-sdk[python]

# Connect to the federation
from defoneos_sdk import DefoneosClient

client = DefoneosClient(
    endpoint="https://your-defoneos.example.com/mcp",
    auth_token="your-oauth2-token",  # OAuth2 JWT Ed25519
)

# Discover available tools (30 MCPs × multiple tools each)
tools = client.discover_tools()
print(f"Available tools: {len(tools)}")
# → Available tools: 142

# Call a sensor fusion tool
result = client.call("defoneos-fusion-mcp", "fuse_isr_feed", {
    "sources": ["sentinel2", "ais", "rtsp_cam_1"],
    "bbox": [-1.5, 53.7, -1.2, 53.9"],  # Yorkshire
    "time_window": "PT1H",
})
print(result.fused_track_count)  # → 47 tracks

# Subscribe to SIGIL audit events
for event in client.stream("sigil.events", filter={"action": "drone_command"}):
    print(f"[SIGIL] {event.actor} → {event.action}: {event.digest[:16]}...")

TypeScript — Connect with OAuth2

// npm install @defoneos/sdk
import { DefoneosClient } from '@defoneos/sdk';

const client = new DefoneosClient({
  endpoint: 'https://your-defoneos.example.com/mcp',
  auth: {
    type: 'oauth2',
    clientId: process.env.DEFONEOS_CLIENT_ID,
    clientSecret: process.env.DEFONEOS_CLIENT_SECRET,
    scopes: ['fusion:read', 'swarm:command', 'sigil:audit'],
  },
});

// List all MCP servers in the federation
const servers = await client.listServers();
console.log(`Federation: ${servers.length} MCP servers online`);

// Command a drone swarm (requires swarm:command scope)
const mission = await client.call('defoneos-swarm-mcp', 'launch_patrol', {
  drones: ['drone-01', 'drone-02', 'drone-03'],
  pattern: 'grid_search',
  area: { lat: 53.8, lng: -1.3, radius_km: 5 },
  altitude_m: 120,
  sigil_audit: true,  // Every command logged to SIGIL chain
});

The REST API exposes all DEFONEOS capabilities over standard HTTP/2. Use this for web dashboards, third-party system integration, and non-AI workloads.

Authentication

# Step 1: Get OAuth2 token
curl -X POST https://your-defoneos.example.com/auth/token \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=client_credentials" \
  -d "client_id=$CLIENT_ID" \
  -d "client_secret=$CLIENT_SECRET" \
  -d "scope=fusion:read swarm:command sigil:audit"

# Response:
# {"access_token":"eyJ...","token_type":"Bearer","expires_in":3600}

# Step 2: Use token for API calls
curl https://your-defoneos.example.com/api/v1/fusion/tracks \
  -H "Authorization: Bearer eyJ..." \
  -H "Accept: application/json"

Key REST Endpoints

Rate Limits

Subscribe to real-time event streams for live updates — SIGIL events, drone telemetry, alerts, and fusion track changes.

Python WebSocket Client

import websockets
import asyncio
import json

async def defoneos_stream():
    uri = "wss://your-defoneos.example.com/ws/v1/stream"
    async with websockets.connect(
        uri,
        extra_headers={"Authorization": f"Bearer {token}"}
    ) as ws:
        # Subscribe to channels
        await ws.send(json.dumps({
            "action": "subscribe",
            "channels": ["sigil.events", "fusion.tracks", "swarm.telemetry", "alerts.critical"]
        }))

        async for message in ws:
            event = json.loads(message)
            if event["channel"] == "alerts.critical":
                print(f"🚨 ALERT: {event['data']['title']}")
            elif event["channel"] == "sigil.events":
                print(f"[SIGIL] {event['data']['digest'][:16]} {event['data']['action']}")

asyncio.run(defoneos_stream())

Available WebSocket Channels

DEFONEOS includes a native C2 bridge that speaks the TAK protocol — making it compatible with ATAK, WinTAK, iTAK, and FreeTAKServer out of the box.

Architecture

┌──────────────┐     CoT XML      ┌──────────────┐    MCP JSON-RPC   ┌──────────────┐
│  ATAK Client │ ←──────────────→ │ FreeTAKServer│ ←───────────────→ │  DEFONEOS    │
│  (Android)   │    (TAK Proto)   │   (Bridge)   │  (Bidirectional)  │  MCP Suite   │
└──────────────┘                  └──────────────┘                   └──────────────┘
       ↑                                                                ↑
       │ CoT display                                                    │ ISR Fusion
       │ (tracks, alerts)                                               │ Swarm control
       │                                                                │ SIGIL audit
┌──────┴──────┐                                                  ┌──────┴──────┐
│  ATAK User  │                                                  │  Cesium COP │
│  (Operator) │                                                  │  (3D Map)   │
└─────────────┘                                                  └─────────────┘

Configure FreeTAKServer → DEFONEOS Bridge

# /opt/freetakserver/config/defoneos_bridge.yml
defoneos:
  endpoint: "https://your-defoneos.example.com/mcp"
  auth_token: "${DEFONEOS_TOKEN}"
  
  # Routes: which CoT types to forward to DEFONEOS
  cot_routes:
    - type: "a-f-A"          # Friendly air track
      forward_to: "defoneos-fusion-mcp.ingest_cot"
    - type: "a-h-A"          # Friendly ground track
      forward_to: "defoneos-fusion-mcp.ingest_cot"
    - type: "b-m-r"          # Emergency alert
      forward_to: "defoneos-sigil-mcp.emit_alert"
  
  # Reverse: DEFONEOS fusion tracks → CoT for ATAK display
  fusion_to_cot:
    enabled: true
    min_confidence: 0.7
    cot_type_template: "a-f-A-C-M"  # Friend, Air, Combat, Manual
    sigil_signed: true            # Attach SIGIL digest to each CoT

For multi-organisation deployments (e.g., joint operations, coalition partners), DEFONEOS supports SIGIL chain synchronisation — allowing each party to maintain a verifiable, tamper-evident audit trail.

Python — SIGIL Chain Subscriber

from defoneos_sdk import SigilSync

# Connect to partner's SIGIL chain as read-only auditor
partner_chain = SigilSync(
    endpoint="https://partner-defoneos.example.com/api/v1/sigil",
    auth_token="coalition-audit-token",
    mode="read_only",  # We can verify, not write
)

# Verify the entire chain integrity
report = partner_chain.verify_chain()
print(f"Chain entries: {report.entry_count}")
print(f"Hash chain valid: {report.hash_chain_valid}")
print(f"Signatures valid: {report.signatures_valid}")
print(f"Tamper detected: {report.tamper_detected}")

# Subscribe to new SIGIL events (live audit feed)
async for entry in partner_chain.stream():
    # Verify each new entry's signature independently
    assert entry.verify_signature(), "Invalid signature!"
    assert entry.verify_hash_link(), "Hash chain broken!"
    log_audit(entry)

SIGIL Entry Structure

{
  "seq": 47823,
  "timestamp": "2026-07-06T07:45:01.234Z",
  "actor": "swarm-mcp",
  "action": "launch_patrol",
  "payload_hash": "sha256:a1b2c3...",
  "prev_digest": "sha256:9z8y7...",
  "digest": "sha256:e5d4f3...",
  "signature": "ed25519:8b7a6...",
  "signing_key": "ed25519:pub:11a2b..."
}