acp-protocol 1.0.0
Agent Communication Protocol (ACP) - A standardized protocol for multi-agent system communication
Requires Python: >=3.10
Keywords: agents, ai, communication, llm, multi-agent, protocol

# ACP - Agent Communication Protocol

[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![Protocol Version](https://img.shields.io/badge/ACP-v1.0.0-green.svg)](https://github.com/acp-protocol/acp-python)

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

```bash
pip install acp-protocol
```

With LangChain integration:

```bash
pip install acp-protocol[langchain]
```

For development:

```bash
pip install acp-protocol[dev]
```

## Quick Start

### Creating Messages

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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:

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

## Testing

```bash
# 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:

```python
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](LICENSE) for details.

## Acknowledgments

Based on the [ACP v1.0 Specification](https://github.com/acp-protocol/specification) by the ACP Working Group.

## Links

- [ACP Specification](https://github.com/acp-protocol/specification)
- [GitHub Repository](https://github.com/acp-protocol/acp-python)
- [Issue Tracker](https://github.com/acp-protocol/acp-python/issues)