acp-protocol 1.0.0

Agent Communication Protocol (ACP) - A standardized protocol for multi-agent system communication

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:

Intent	Description	Required Content
REQUEST	Ask for information	content.query
DELEGATE	Assign a task	content.task
VERIFY	Check correctness	content.target_message_id
CHALLENGE	Dispute a claim	content.reason
PROPOSE	Suggest action	content.proposal
ACCEPT	Agree to proposal	parent_message_id
REJECT	Decline proposal	content.reason
ESCALATE	Raise to supervisor	content.reason
REPORT	Status update	content.status
TERMINATE	End interaction	content.reason

Trust Levels

Level	Description	Score
VERIFIED	Cryptographically proven	1.0
TRUSTED	Good historical performance	0.8
NEUTRAL	Unknown/new agent	0.5
SUSPICIOUS	Warning flags present	0.2
BLOCKED	Do not interact	0.0

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