Cross-Organizational Multi-Agent Systems: Federated Agents Across Company Boundaries

When Agents Cross Company Boundaries

Multi-agent systems become significantly more complex when agents from different organizations need to collaborate. A supply chain optimization system might involve a manufacturer's demand forecasting agent, a logistics provider's routing agent, and a retailer's inventory management agent — each owned by a different company with different data policies, security requirements, and business objectives.

This is not a theoretical concern. As AI agent ecosystems mature, federated multi-agent architectures are becoming necessary for any workflow that spans organizational boundaries. The challenge is building trust, enforcing data boundaries, and maintaining compliance when you do not control the other side.

Trust Boundaries and the Agent Gateway

The first principle of cross-organizational agent design: never trust the other organization's agents directly. All communication goes through a gateway that validates, sanitizes, and logs every interaction.

flowchart TD
 INPUT(["Task input"])
 SUPER["Supervisor agent<br/>plans plus monitors"]
 W1["Worker 1<br/>research"]
 W2["Worker 2<br/>code"]
 W3["Worker 3<br/>writing"]
 CRITIC{"Output meets<br/>rubric?"}
 REWORK["Rework or<br/>retry path"]
 SHARED[("Shared scratchpad<br/>and memory")]
 OUT(["Final result"])
 INPUT --> SUPER
 SUPER --> W1 --> CRITIC
 SUPER --> W2 --> CRITIC
 SUPER --> W3 --> CRITIC
 W1 --> SHARED
 W2 --> SHARED
 W3 --> SHARED
 SHARED --> SUPER
 CRITIC -->|Pass| OUT
 CRITIC -->|Fail| REWORK --> SUPER
 style SUPER fill:#4f46e5,stroke:#4338ca,color:#fff
 style CRITIC fill:#f59e0b,stroke:#d97706,color:#1f2937
 style OUT fill:#059669,stroke:#047857,color:#fff
 style SHARED fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any
import hashlib
import json

class TrustLevel(Enum):
 UNTRUSTED = "untrusted"
 BASIC = "basic"
 VERIFIED = "verified"
 PRIVILEGED = "privileged"

@dataclass
class OrganizationProfile:
 org_id: str
 name: str
 trust_level: TrustLevel
 allowed_operations: list[str]
 data_classification_ceiling: str # "public", "internal", "confidential"
 rate_limit_per_minute: int = 60
 api_key_hash: str = ""

@dataclass
class AgentMessage:
 source_org: str
 source_agent: str
 target_org: str
 target_agent: str
 operation: str
 payload: dict[str, Any]
 timestamp: str = field(
 default_factory=lambda: datetime.now().isoformat()
 )
 message_id: str = ""

class AgentGateway:
 def __init__(self, own_org_id: str):
 self.own_org_id = own_org_id
 self.org_registry: dict[str, OrganizationProfile] = {}
 self.audit_log: list[dict] = []

 def register_org(self, profile: OrganizationProfile):
 self.org_registry[profile.org_id] = profile

 def process_inbound(self, message: AgentMessage) -> dict:
 org = self.org_registry.get(message.source_org)
 if not org:
 return self._reject("Unknown organization", message)

 if org.trust_level == TrustLevel.UNTRUSTED:
 return self._reject("Organization not trusted", message)

 if message.operation not in org.allowed_operations:
 return self._reject(
 f"Operation '{message.operation}' not permitted",
 message)

 sanitized = self._sanitize_payload(message.payload, org)

 self._audit(message, "accepted")
 return {
 "status": "accepted",
 "sanitized_payload": sanitized,
 "trust_level": org.trust_level.value,
 }

 def process_outbound(
 self, message: AgentMessage, data_classification: str
 ) -> dict:
 org = self.org_registry.get(message.target_org)
 if not org:
 return self._reject("Unknown target org", message)

 classification_rank = {
 "public": 0, "internal": 1, "confidential": 2
 }
 if classification_rank.get(data_classification, 99) > classification_rank.get(org.data_classification_ceiling, 0):
 return self._reject(
 f"Data classification '{data_classification}' exceeds "
 f"ceiling '{org.data_classification_ceiling}'",
 message)

 filtered = self._filter_outbound_data(
 message.payload, org.data_classification_ceiling
 )
 self._audit(message, "sent")
 return {"status": "sent", "filtered_payload": filtered}

 def _sanitize_payload(self, payload: dict, org) -> dict:
 sanitized = {}
 for key, value in payload.items():
 if isinstance(value, str) and len(value) > 10000:
 sanitized[key] = value[:10000]
 else:
 sanitized[key] = value
 return sanitized

 def _filter_outbound_data(self, payload, ceiling):
 return {k: v for k, v in payload.items()
 if not k.startswith("_internal")}

 def _reject(self, reason, message):
 self._audit(message, f"rejected: {reason}")
 return {"status": "rejected", "reason": reason}

 def _audit(self, message, action):
 self.audit_log.append({
 "timestamp": datetime.now().isoformat(),
 "source_org": message.source_org,
 "target_org": message.target_org,
 "operation": message.operation,
 "action": action,
 })

API Contracts for Agent Interoperability

When two organizations agree to let their agents communicate, they need a formal contract defining the operations, data schemas, SLAs, and failure modes.

@dataclass
class AgentAPIContract:
 contract_id: str
 party_a: str
 party_b: str
 operations: list[dict] # name, request_schema, response_schema
 sla: dict # max_latency_ms, availability_percent, etc.
 data_policy: dict # retention, allowed_fields, redacted_fields
 effective_date: str
 expiration_date: str

 def validate_request(self, operation: str, payload: dict) -> dict:
 op_spec = next(
 (o for o in self.operations if o["name"] == operation),
 None)
 if not op_spec:
 return {"valid": False, "error": "Operation not in contract"}

 required_fields = op_spec.get("request_schema", {}).get(
 "required", []
 )
 missing = [f for f in required_fields if f not in payload]
 if missing:
 return {
 "valid": False,
 "error": f"Missing fields: {missing}",
 }

 redacted = self.data_policy.get("redacted_fields", [])
 for field_name in redacted:
 if field_name in payload:
 return {
 "valid": False,
 "error": f"Field '{field_name}' must not be sent",
 }

 return {"valid": True}

# Define a contract between two organizations
supply_chain_contract = AgentAPIContract(
 contract_id="SC-2026-001",
 party_a="manufacturer_co",
 party_b="logistics_co",
 operations=[
 {
 "name": "request_shipping_quote",
 "request_schema": {
 "required": ["origin", "destination", "weight_kg"],
 },
 "response_schema": {
 "required": ["quote_id", "price_usd", "eta_days"],
 },
 },
 {
 "name": "track_shipment",
 "request_schema": {"required": ["tracking_id"]},
 "response_schema": {
 "required": ["status", "current_location"],
 },
 },
 ],
 sla={"max_latency_ms": 5000, "availability_percent": 99.5},
 data_policy={
 "retention_days": 90,
 "redacted_fields": ["customer_ssn", "internal_cost"],
 },
 effective_date="2026-01-01",
 expiration_date="2026-12-31")

Data Sharing With Privacy Controls

Cross-organizational data sharing requires explicit controls over what data leaves your boundary, how it is transformed, and what the receiving party can do with it.

class DataSharingController:
 def __init__(self):
 self.sharing_rules: dict[str, dict] = {}

 def add_rule(
 self, target_org: str, allowed_fields: list[str],
 transforms: dict[str, str] | None = None):
 self.sharing_rules[target_org] = {
 "allowed_fields": set(allowed_fields),
 "transforms": transforms or {},
 }

 def prepare_for_sharing(
 self, data: dict, target_org: str
 ) -> dict:
 rule = self.sharing_rules.get(target_org)
 if not rule:
 return {} # share nothing by default

 filtered = {
 k: v for k, v in data.items()
 if k in rule["allowed_fields"]
 }

 for field_name, transform_type in rule["transforms"].items():
 if field_name in filtered:
 filtered[field_name] = self._apply_transform(
 filtered[field_name], transform_type
 )

 return filtered

 def _apply_transform(self, value, transform_type: str):
 if transform_type == "hash":
 return hashlib.sha256(str(value).encode()).hexdigest()[:16]
 elif transform_type == "round":
 return round(float(value), 0)
 elif transform_type == "redact":
 return "[REDACTED]"
 return value

# Only share specific fields, with transforms for sensitive values
controller = DataSharingController()
controller.add_rule(
 target_org="logistics_co",
 allowed_fields=["order_id", "weight_kg", "destination_zip", "customer_id"],
 transforms={"customer_id": "hash"}, # pseudonymize
)

Compliance and Audit Trail

Every cross-organizational interaction must be auditable. Regulations like GDPR, HIPAA, and require proof of what data was shared, with whom, and under what authority. The audit log in the gateway provides this, but you should also maintain a compliance checker that validates ongoing adherence to contracts and policies.

FAQ

How do I handle version mismatches when the other organization updates their agent API?

Use semantic versioning in your API contracts and support at least the current and previous major version simultaneously. Include a version field in every agent message. The gateway should reject messages with unsupported versions and log them for debugging. Negotiate upgrade timelines in your contract — typically 90 days of overlap between versions.

What happens when a cross-organizational agent call fails or times out?

Design for failure at every level. Set aggressive timeouts (the SLA max latency), implement circuit breakers that stop calling a failing external agent after 3 consecutive failures, and always have a local fallback. For a shipping quote, the fallback might be a cached recent quote or an estimated range. Never let an external agent failure cascade into your internal system going down.

How do I verify that the other organization's agent is not sending manipulated data?

Use cryptographic signing for all inter-organizational messages. Each organization signs outbound messages with its private key, and the receiving gateway verifies the signature. For high-stakes operations, add a mutual attestation step where both parties agree on the message contents before either acts on them. This prevents replay attacks and tampered payloads.

---

#FederatedAgents #CrossOrganization #APIContracts #TrustBoundaries #Python #AgenticAI #LearnAI #AIEngineering