🔐 A2SPA Developer Documentation

🖊️ Sending Messages

To send a message (payload), you need to create it, hash it, sign it, and send it to A2SPA for verification. Here's how it works:

💡 4 Simple Steps

1. Create a message with sender and receiver IDs, timestamp, and a unique code (nonce).
2. Hash the canonical signable payload to create a unique fingerprint.
3. Sign those exact same canonical bytes with your private key.
4. Send the message and signature to A2SPA's API, then poll the receiver inbox if you want to read delivered messages through the API.

🔧 Complete Example

What You Need:

• Two agents: one sender, one receiver.
• The sender's private key (saved in a .priv.pem file).
• Your API key from the dashboard.

🐍 Python Code to Send a Message

import json, uuid, requests, hashlib, re
from datetime import datetime, timezone
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding

# ============ Settings ============
API_BASE = "https://aimodularity.com/A2SPA"
API_KEY = "YOUR_API_KEY_HERE"  # Get from dashboard
USER_ID = "your_user_id"
SENDER_AGENT_ID = f"{USER_ID}_my_agent_send"
RECEIVER_AGENT_ID = f"{USER_ID}_my_agent_receive"
PRIVATE_KEY_PATH = "your_sender_agent.priv.pem"
SIGNABLE_FIELDS = (
    "agent_id",
    "target_agent_id",
    "timestamp",
    "nonce",
    "input",
    "output",
    "alert_threshold",
)
EXPECTED_HASH_RE = re.compile(r"expected:\s*([0-9a-f]{64})", re.IGNORECASE)

# ============ Helper Function ============
def normalize_a2spa_payload(payload):
    normalized = {field: payload[field] for field in SIGNABLE_FIELDS if field in payload}
    normalized.setdefault("alert_threshold", 10)
    return normalized


def canonicalize_a2spa_payload(payload):
    return json.dumps(normalize_a2spa_payload(payload), sort_keys=True)


def prepare_signed_bytes(payload):
    return canonicalize_a2spa_payload(payload).encode("utf-8")


def compute_payload_hash(payload):
    return hashlib.sha256(prepare_signed_bytes(payload)).hexdigest()


def finalize_payload(payload_before_hash):
    final_payload = {k: v for k, v in payload_before_hash.items() if k not in ("hash", "signature")}
    final_payload["hash"] = compute_payload_hash(payload_before_hash)
    return final_payload


def sign_payload(private_key, payload_before_hash):
    return private_key.sign(
        prepare_signed_bytes(payload_before_hash),
        padding.PKCS1v15(),
        hashes.SHA256(),
    ).hex()


def build_request_body(payload, signature):
    return {"payload": payload, "signature": signature}


def extract_expected_hash(error_text):
    match = EXPECTED_HASH_RE.search(str(error_text or ""))
    return match.group(1) if match else None

# ============ Step 1: Create Message ============
payload_before_hash = {
    "agent_id": SENDER_AGENT_ID,
    "target_agent_id": RECEIVER_AGENT_ID,
    "timestamp": datetime.now(timezone.utc).isoformat(),
    "nonce": str(uuid.uuid4()),  # Unique code to prevent reuse
    "input": {"message": "Hello from sender agent!"},
    "output": {"status": "ready"},
    "alert_threshold": 10,
}
print("✅ Step 1: Message created")

# ============ Step 2: Hash the Message ============
payload = finalize_payload(payload_before_hash)
print(f"✅ Step 2: Hash created: {payload['hash'][:16]}...")

# ============ Step 3: Sign the Message ============
with open(PRIVATE_KEY_PATH, "rb") as f:
    PRIVATE_KEY_PEM = f.read()

private_key = serialization.load_pem_private_key(
    PRIVATE_KEY_PEM, 
    password=None
)

signature_hex = sign_payload(private_key, payload_before_hash)
print(f"✅ Step 3: Signature created: {signature_hex[:16]}...")

# ============ Step 4: Send to A2SPA ============
request_body = build_request_body(payload, signature_hex)
resp = requests.post(
    f"{API_BASE}/api/verify_payload",
    headers={
        "Content-Type": "application/json",
        "x-api-key": API_KEY
    },
    json=request_body
)

print(f"\n🎯 Result:")
print(f"Status Code: {resp.status_code}")
print(f"Response: {resp.text}")

if resp.status_code == 200:
    print("\n✅ Success! Message verified and logged.")
else:
    debug_snapshot = {
        "payload_before_hash": payload_before_hash,
        "canonical_string": canonicalize_a2spa_payload(payload_before_hash),
        "canonical_bytes_hex": prepare_signed_bytes(payload_before_hash).hex(),
        "client_computed_hash": compute_payload_hash(payload_before_hash),
        "final_request_body": request_body,
        "server_expected_hash": extract_expected_hash(resp.text),
    }
    print("\n❌ Hash/signature debug:")
    print(json.dumps(debug_snapshot, indent=2, sort_keys=True))

🟨 JavaScript / TypeScript Starter Behavior

The generated JavaScript and TypeScript starters now build payload_before_hash, compute payload.hash from the verifier-normalized payload, sign the exact same canonical UTF-8 bytes, and then send { payload, signature }.

const input = { message: "Hello from my existing Node agent" };
const output = await existingAgent(input);
const signedRequest = await client.buildSignedRequest(
  process.env.A2SPA_TARGET_AGENT_ID,
  input,
  output,
);

const firstResponse = await client.postRequestBody(signedRequest.requestBody);
console.log(`First request: ${firstResponse.status} ${await firstResponse.text()}`);

const secondResponse = await client.postRequestBody(signedRequest.requestBody);
console.log(`Second request: ${secondResponse.status} ${await secondResponse.text()}`);

// Expected live behavior:
// First request: 200 {"success":true}
// Second request: 409 {"error":"Replay attack detected - nonce already used"}

Avoid global string replacements such as .replace(/:/g, ": ") or .replace(/,/g, ", "). They corrupt timestamps and other string values instead of formatting only JSON separators.

📥 Polling a Receiver Inbox

curl -H "x-api-key: YOUR_API_KEY_HERE" "https://aimodularity.com/A2SPA/api/inbox_for_agent?agent_id=YOUR_RECEIVER_AGENT_ID&limit=25"

Use the inbox endpoint when an existing agent needs to poll for received messages through the A2SPA API. Use /api/logs_for_agent when you want audit history or troubleshooting detail.

📋 Message Fields

🧮 Canonicalization Contract

The verifier hashes and signs the schema-normalized payload, not the raw request body. The current contract matches Python json.dumps(payload, sort_keys=True) over the signable payload fields.

⚠️ Common Errors

🧭 Delegation & Policy Boundary

on_behalf_of is never authorization by itself. It is metadata unless backed by a verified delegation proof.

It is not trusted unless backed by a verified delegation proof.

Cryptographic binding proves integrity. Policy determines semantic permission.

No signature. No execution.

• Delegated execution is not fully implemented in the current verifier flow.
• A future delegation design must require a second signature, capability token, or explicit delegation policy object.
• Intent/action binding is currently byte-level integrity binding.
• A2SPA proves action, target, and input bytes were not changed after signing.
• Semantic compatibility between intent and action belongs in explicit policy or schema rules, not in the current integrity check alone.