acp-protocol

ACP - Agent Communication Protocol

A standardized, enforceable protocol for communication between AI agents in multi-agent systems.

Why ACP?

Multi-agent systems today suffer from:

• Unstructured Communication: Agents pass free-form text or inconsistent JSON with no semantic guarantees
• No Confidence Modeling: Agents cannot express uncertainty or enable weighted consensus
• Constraint Violations: No formal way to specify or enforce constraints
• Trust Failures: No agent reputation system or capability verification
• Debugging Impossibility: Post-mortem analysis requires archaeology

ACP solves these problems with:

• Standardized message schema with Pydantic validation
• First-class confidence with calibration tracking
• Enforceable constraints with validation hooks
• Trust levels with automatic degradation/upgrade
• Structured intents for semantic routing

Installation

pip install acp-protocol

With LangChain integration:

pip install acp-protocol[langchain]

For development:

pip install acp-protocol[dev]

Quick Start

Creating Messages

from acp import ACPMessage, Intent, Constraints, Context

# Create a delegation message
message = ACPMessage(
    sender_id="coordinator_agent",
    recipient_id="summarizer_agent",
    intent=Intent.DELEGATE,
    content={
        "task": "summarize_document",
        "document_id": "doc_123",
    },
    confidence=0.87,
    constraints=Constraints(
        max_tokens=800,
        deadline_ms=5000,
    ),
    context=Context(
        domain="legal",
        urgency="high",
    ),
)

# Create a response
response = message.create_report(
    sender_id="summarizer_agent",
    status="completed",
    confidence=0.92,
    result={"summary": "The document discusses..."},
)

Validating Messages

from acp import ACPValidator

validator = ACPValidator(strict_mode=True)

# Validate and raise on error
validated_message = validator.validate_or_raise(message)

# Or get detailed result
result = validator.validate(message)
if not result.valid:
    for error in result.errors:
        print(f"Error: {error}")

Confidence Calibration

from acp import Calibrator

calibrator = Calibrator()

# Record outcomes after tasks complete
calibrator.record_outcome("agent_01", claimed_confidence=0.90, success=True)
calibrator.record_outcome("agent_01", claimed_confidence=0.85, success=False)

# Adjust future confidence claims based on history
adjusted = calibrator.adjust_confidence("agent_01", claimed_confidence=0.90)
# Returns lower value if agent is historically overconfident

# Get calibration metrics
metrics = calibrator.get_metrics("agent_01")
print(f"Reliability: {metrics.reliability_score:.2%}")
print(f"ECE: {metrics.expected_calibration_error:.4f}")

Trust Management

from acp import TrustManager, TrustLevel

trust_manager = TrustManager()

# Record successes/failures
trust_manager.record_success("agent_01")
trust_manager.record_failure("agent_02", "Task timeout")

# Check trust level
level = trust_manager.get_trust_level("agent_01")
print(f"Trust level: {level.value}")  # "trusted"

# Block problematic agents
trust_manager.block_agent("bad_agent", "Malicious behavior")

Routing

from acp import Router, AgentCapability, Calibrator, TrustManager

router = Router()
calibrator = Calibrator()
trust_manager = TrustManager()

# Register agents with capabilities
router.register_agent(AgentCapability(
    agent_id="summarizer_01",
    capabilities={"summarization", "text_processing"},
    min_deadline_ms=1000,
    max_tokens=2000,
))

# Route to best agent
best_agent = router.route(
    message,
    calibrator=calibrator,
    trust_manager=trust_manager,
    required_capabilities={"summarization"},
)

LangChain Integration

from acp.adapters import ACPLangChainAdapter
from acp import Intent

adapter = ACPLangChainAdapter(
    agent_id="langchain_agent",
    default_domain="general",
)

# Create delegation
delegation = adapter.create_delegation(
    task="Summarize this document",
    recipient_id="summarizer",
    confidence=0.85,
)

# Wrap LangChain output as ACP message
response = adapter.wrap_langchain_response(
    langchain_output=chain_result,
    recipient_id="coordinator",
    intent=Intent.REPORT,
    confidence=0.90,
)

# Unwrap ACP message for LangChain
chain_input = adapter.create_chain_input(incoming_message)

Core Concepts

Intents

ACP defines 10 semantic intents:

Trust Levels

Constraints

Constraints(
    max_tokens=800,          # Maximum response length
    deadline_ms=5000,        # Response deadline (≥100ms)
    quality_threshold=0.85,  # Minimum quality score
    max_cost_usd=0.50,       # Budget limit
    require_sources=True,    # Must cite sources
    allowed_tools=["search", "calculator"],  # Permitted tools
)

Confidence

• Range: [0.01, 0.99] (no perfect certainty allowed)
• Optional uncertainty bounds: (lower, upper) tuple
• Calibration system tracks accuracy over time

API Reference

Models

• ACPMessage - Core message structure
• Intent - Message intent enum
• TrustLevel - Trust level enum
• Constraints - Task constraints
• Context - Message context

Validation

• ACPValidator - Multi-layer message validator
• ValidationResult - Validation outcome

Calibration

• Calibrator - Confidence calibration system
• CalibrationMetrics - Agent calibration data

Trust

• TrustManager - Agent trust management
• TrustEvent - Trust change record

Routing

• Router - Capability-based routing
• AgentCapability - Agent capability description

Adapters

• BaseACPAdapter - Abstract adapter base class
• ACPLangChainAdapter - LangChain integration

Converters

• UnstructuredToACPConverter - Legacy text to ACP

Exceptions

• ACPError - Base exception
• ValidationError - Validation failures
• IntentValidationError - Intent/content mismatch
• ConstraintValidationError - Invalid constraints
• TemporalValidationError - Timestamp/deadline issues
• TrustValidationError - Trust violations
• RoutingError - No suitable agent found

Examples

See the examples/ directory:

• basic_usage.py - Fundamental operations
• delegation_flow.py - Multi-agent workflow
• langchain_example.py - LangChain integration

Run examples:

python examples/basic_usage.py
python examples/delegation_flow.py
python examples/langchain_example.py

Testing

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# With coverage
pytest --cov=src/acp --cov-report=term-missing

Protocol Version

This library implements ACP v1.0.0.

Protocol version is included in every message:

message.protocol_version  # "1.0.0"

Contributing

Contributions are welcome! Please:

1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Submit a pull request

License

MIT License - see LICENSE for details.

Acknowledgments

Based on the ACP v1.0 Specification by the ACP Working Group.

Links

• ACP Specification
• GitHub Repository
• Issue Tracker