Compare commits

...

1 Commits

Author SHA1 Message Date
HAL9000 55732df282 docs: add A2A protocol and implementation guide
- Add comprehensive A2A Protocol Specification (docs/reference/a2a_protocol_spec.md)
  * JSON-RPC 2.0 foundation and extension methods
  * Session, plan, agent, skill, and context management methods
  * Message types, formats, and transport details
  * Error handling with standard and custom error codes
  * Protocol examples and versioning
  * Security considerations

- Add A2A Implementation Guide (docs/guides/a2a_implementation_guide.md)
  * Stdio transport implementation for local mode
  * HTTP transport implementation for server mode
  * LangGraph Platform RemoteGraph integration
  * Session and plan lifecycle management
  * Error recovery strategies and best practices

- Add A2A Server Deployment Guide (docs/guides/a2a_server_deployment.md)
  * Docker and Docker Compose deployment
  * Kubernetes deployment with manifests
  * Helm chart configuration and installation
  * Environment variables and security configuration
  * Monitoring, logging, and operational best practices

These guides provide comprehensive documentation for implementing and deploying
A2A Protocol in production environments, addressing the core server implementation
requirements for milestone M9.
2026-04-19 13:28:06 +00:00
3 changed files with 2302 additions and 0 deletions
+682
View File
@@ -0,0 +1,682 @@
# A2A Implementation Guide
## Overview
This guide provides comprehensive instructions for implementing the Agent-to-Agent (A2A) Protocol in CleverAgents applications. It covers both local (stdio) and remote (HTTP) transport mechanisms, integration with LangGraph Platform, and best practices for production deployments.
## A2A Implementation Overview
The A2A Protocol implementation consists of several key components:
1. **Protocol Handler**: Manages JSON-RPC 2.0 message parsing and routing
2. **Transport Layer**: Handles communication over stdio or HTTP
3. **Session Manager**: Manages agent sessions and lifecycle
4. **Plan Executor**: Executes and monitors plans
5. **Error Handler**: Manages error recovery and retry logic
### Architecture
```
┌─────────────────────────────────────────────────────────┐
│ Application Layer │
│ (Agent Logic, Skills, Tools) │
└──────────────────┬──────────────────────────────────────┘
┌──────────────────▼──────────────────────────────────────┐
│ A2A Protocol Handler │
│ (JSON-RPC 2.0 Parsing, Method Routing) │
└──────────────────┬──────────────────────────────────────┘
┌──────────────────▼──────────────────────────────────────┐
│ Session Manager │
│ (Session Lifecycle, Context Management) │
└──────────────────┬──────────────────────────────────────┘
┌──────────┴──────────┐
│ │
┌───────▼────────┐ ┌────────▼──────────┐
│ Stdio Transport│ │ HTTP Transport │
│ (Local Mode) │ │ (Server Mode) │
└────────────────┘ └───────────────────┘
```
## Stdio Transport (Local Mode)
Stdio transport is used for local, in-process agent communication.
### Implementation
#### Message Format
The stdio transport uses a length-prefixed message format:
```
<message_length_in_bytes>\n
<json_message>
```
#### Example Implementation (Python)
```python
import json
import sys
from typing import Dict, Any
class StdioTransport:
def __init__(self):
self.input_stream = sys.stdin.buffer
self.output_stream = sys.stdout.buffer
def send_message(self, message: Dict[str, Any]) -> None:
"""Send a message over stdio."""
json_str = json.dumps(message)
json_bytes = json_str.encode('utf-8')
length = len(json_bytes)
# Send length prefix
self.output_stream.write(f"{length}\n".encode('utf-8'))
# Send message
self.output_stream.write(json_bytes)
self.output_stream.flush()
def receive_message(self) -> Dict[str, Any]:
"""Receive a message over stdio."""
# Read length prefix
length_line = self.input_stream.readline().decode('utf-8').strip()
length = int(length_line)
# Read message
message_bytes = self.input_stream.read(length)
message_str = message_bytes.decode('utf-8')
return json.loads(message_str)
```
#### Example Implementation (JavaScript/Node.js)
```javascript
const readline = require('readline');
class StdioTransport {
constructor() {
this.rl = readline.createInterface({
input: process.stdin,
output: process.stdout,
terminal: false
});
this.buffer = '';
}
sendMessage(message) {
const jsonStr = JSON.stringify(message);
const jsonBytes = Buffer.from(jsonStr, 'utf-8');
const length = jsonBytes.length;
// Send length prefix and message
process.stdout.write(`${length}\n`);
process.stdout.write(jsonBytes);
}
async receiveMessage() {
return new Promise((resolve) => {
this.rl.once('line', (lengthLine) => {
const length = parseInt(lengthLine);
let data = '';
const onData = (chunk) => {
data += chunk.toString();
if (Buffer.byteLength(data, 'utf-8') >= length) {
process.stdin.removeListener('data', onData);
const message = JSON.parse(data.substring(0, length));
resolve(message);
}
};
process.stdin.on('data', onData);
});
});
}
}
```
### Local Mode Configuration
```yaml
# config.yaml
a2a:
transport: stdio
mode: local
session:
timeout: 3600
max_concurrent: 10
error_handling:
retry_attempts: 3
retry_backoff: exponential
retry_delay_ms: 1000
```
## HTTP Transport (Server Mode)
HTTP transport is used for remote, network-based agent communication.
### Implementation
#### Server Setup (Python/FastAPI)
```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import json
from typing import Dict, Any
app = FastAPI()
class JSONRPCRequest(BaseModel):
jsonrpc: str = "2.0"
method: str
params: Dict[str, Any] = {}
id: str
class JSONRPCResponse(BaseModel):
jsonrpc: str = "2.0"
result: Dict[str, Any] = None
error: Dict[str, Any] = None
id: str
class A2AServer:
def __init__(self):
self.sessions = {}
self.handlers = {
"_cleveragents/session/create": self.handle_session_create,
"_cleveragents/session/close": self.handle_session_close,
"_cleveragents/plan/create": self.handle_plan_create,
"_cleveragents/plan/execute": self.handle_plan_execute,
"_cleveragents/plan/status": self.handle_plan_status,
}
async def handle_rpc_call(self, request: JSONRPCRequest) -> JSONRPCResponse:
"""Handle a JSON-RPC 2.0 call."""
try:
handler = self.handlers.get(request.method)
if not handler:
return JSONRPCResponse(
id=request.id,
error={
"code": -32601,
"message": "Method not found"
}
)
result = await handler(request.params)
return JSONRPCResponse(
id=request.id,
result=result
)
except Exception as e:
return JSONRPCResponse(
id=request.id,
error={
"code": -32603,
"message": "Internal error",
"data": {"error": str(e)}
}
)
async def handle_session_create(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Create a new session."""
session_id = params.get("session_id")
agent_id = params.get("agent_id")
self.sessions[session_id] = {
"agent_id": agent_id,
"status": "active",
"created_at": "2024-04-19T10:30:00Z"
}
return {
"session_id": session_id,
"created_at": "2024-04-19T10:30:00Z",
"status": "active"
}
async def handle_session_close(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Close a session."""
session_id = params.get("session_id")
if session_id not in self.sessions:
raise ValueError(f"Session {session_id} not found")
del self.sessions[session_id]
return {
"session_id": session_id,
"closed_at": "2024-04-19T10:30:00Z",
"status": "closed"
}
async def handle_plan_create(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Create a new plan."""
plan_id = params.get("plan_id")
session_id = params.get("session_id")
return {
"plan_id": plan_id,
"session_id": session_id,
"status": "created",
"created_at": "2024-04-19T10:30:00Z"
}
async def handle_plan_execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Execute a plan."""
plan_id = params.get("plan_id")
return {
"plan_id": plan_id,
"execution_id": "exec_123",
"status": "executing",
"started_at": "2024-04-19T10:30:00Z"
}
async def handle_plan_status(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Get plan status."""
plan_id = params.get("plan_id")
return {
"plan_id": plan_id,
"status": "completed",
"progress": {
"total_steps": 2,
"completed_steps": 2,
"current_step": None
},
"result": {"status": "success"},
"error": None
}
# Initialize server
a2a_server = A2AServer()
@app.post("/rpc")
async def rpc_endpoint(request: JSONRPCRequest) -> JSONRPCResponse:
"""Main RPC endpoint."""
return await a2a_server.handle_rpc_call(request)
@app.post("/events")
async def events_endpoint(request: JSONRPCRequest) -> JSONRPCResponse:
"""Event notification endpoint."""
# Handle event notifications
return JSONRPCResponse(
id=request.id,
result={"status": "received"}
)
@app.get("/health")
async def health_check():
"""Health check endpoint."""
return {
"status": "healthy",
"timestamp": "2024-04-19T10:30:00Z",
"version": "1.0"
}
```
### Server Mode Configuration
```yaml
# config.yaml
a2a:
transport: http
mode: server
server:
host: 0.0.0.0
port: 8000
ssl:
enabled: true
cert_path: /etc/ssl/certs/server.crt
key_path: /etc/ssl/private/server.key
session:
timeout: 3600
max_concurrent: 100
error_handling:
retry_attempts: 3
retry_backoff: exponential
retry_delay_ms: 1000
```
## LangGraph Platform RemoteGraph Integration
The A2A Protocol integrates with LangGraph Platform's RemoteGraph for distributed graph execution.
### Integration Example
```python
from langgraph.graph import StateGraph
from langgraph.prebuilt import create_react_agent
from langgraph_sdk import RemoteGraph
import httpx
# Create a RemoteGraph client
client = RemoteGraph(
url="http://localhost:8000",
api_key="your-api-key"
)
# Define agent state
class AgentState(TypedDict):
messages: list
session_id: str
plan_id: str
# Create workflow
workflow = StateGraph(AgentState)
async def process_step(state: AgentState):
"""Process a workflow step."""
# Call remote agent via A2A Protocol
response = await client.invoke(
method="_cleveragents/skill/execute",
params={
"session_id": state["session_id"],
"skill_id": "process_data",
"skill_params": {"data": state["messages"]}
}
)
return {
"messages": state["messages"] + [response["result"]]
}
workflow.add_node("process", process_step)
workflow.set_entry_point("process")
# Compile and run
graph = workflow.compile()
result = await graph.ainvoke({
"messages": [],
"session_id": "sess_123",
"plan_id": "plan_456"
})
```
## Session Lifecycle Management
### Session States
```
┌─────────┐
│ Created │
└────┬────┘
┌─────────┐
│ Active │◄──────────┐
└────┬────┘ │
│ │
├─ Suspended ────┤
│ │
▼ │
┌─────────┐ │
│ Closing │───────────┘
└────┬────┘
┌─────────┐
│ Closed │
└─────────┘
```
### Session Management Code
```python
class SessionManager:
def __init__(self):
self.sessions = {}
async def create_session(self, session_id: str, agent_id: str) -> Dict[str, Any]:
"""Create a new session."""
session = {
"session_id": session_id,
"agent_id": agent_id,
"status": "active",
"created_at": datetime.now().isoformat(),
"context": {}
}
self.sessions[session_id] = session
return session
async def get_session(self, session_id: str) -> Dict[str, Any]:
"""Get session by ID."""
if session_id not in self.sessions:
raise ValueError(f"Session {session_id} not found")
return self.sessions[session_id]
async def update_session_context(self, session_id: str, context: Dict[str, Any]) -> None:
"""Update session context."""
session = await self.get_session(session_id)
session["context"].update(context)
async def close_session(self, session_id: str) -> Dict[str, Any]:
"""Close a session."""
session = await self.get_session(session_id)
session["status"] = "closed"
session["closed_at"] = datetime.now().isoformat()
return session
```
## Plan Lifecycle Management
### Plan States
```
┌─────────┐
│ Created │
└────┬────┘
┌──────────┐
│ Queued │
└────┬─────┘
┌──────────┐
│ Executing│◄──────────┐
└────┬─────┘ │
│ │
├─ Paused ────────┤
│ │
├─ Failed ────────┤
│ │
▼ │
┌──────────┐ │
│ Completed├───────────┘
└──────────┘
```
### Plan Execution Code
```python
class PlanExecutor:
def __init__(self):
self.plans = {}
self.executions = {}
async def create_plan(self, plan_id: str, plan_definition: Dict[str, Any]) -> Dict[str, Any]:
"""Create a new plan."""
plan = {
"plan_id": plan_id,
"definition": plan_definition,
"status": "created",
"created_at": datetime.now().isoformat()
}
self.plans[plan_id] = plan
return plan
async def execute_plan(self, plan_id: str, session_id: str) -> Dict[str, Any]:
"""Execute a plan."""
plan = self.plans[plan_id]
execution_id = f"exec_{uuid.uuid4()}"
execution = {
"execution_id": execution_id,
"plan_id": plan_id,
"session_id": session_id,
"status": "executing",
"started_at": datetime.now().isoformat(),
"progress": {
"total_steps": len(plan["definition"]["steps"]),
"completed_steps": 0,
"current_step": None
}
}
self.executions[execution_id] = execution
# Execute steps
for i, step in enumerate(plan["definition"]["steps"]):
execution["progress"]["current_step"] = step["id"]
# Execute step logic here
execution["progress"]["completed_steps"] = i + 1
execution["status"] = "completed"
execution["completed_at"] = datetime.now().isoformat()
return execution
async def get_plan_status(self, plan_id: str) -> Dict[str, Any]:
"""Get plan execution status."""
plan = self.plans[plan_id]
# Find latest execution
executions = [e for e in self.executions.values() if e["plan_id"] == plan_id]
if executions:
return executions[-1]
return {"status": "not_executed"}
```
## Error Recovery
### Retry Strategy
```python
import asyncio
from typing import Callable, Any
class RetryPolicy:
def __init__(self, max_attempts: int = 3, backoff_factor: float = 2.0):
self.max_attempts = max_attempts
self.backoff_factor = backoff_factor
async def execute_with_retry(self, func: Callable, *args, **kwargs) -> Any:
"""Execute a function with retry logic."""
last_exception = None
for attempt in range(self.max_attempts):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < self.max_attempts - 1:
delay = 1.0 * (self.backoff_factor ** attempt)
await asyncio.sleep(delay)
raise last_exception
```
### Circuit Breaker Pattern
```python
from enum import Enum
from datetime import datetime, timedelta
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class CircuitBreaker:
def __init__(self, failure_threshold: int = 5, timeout_seconds: int = 60):
self.failure_threshold = failure_threshold
self.timeout_seconds = timeout_seconds
self.state = CircuitState.CLOSED
self.failure_count = 0
self.last_failure_time = None
async def call(self, func: Callable, *args, **kwargs) -> Any:
"""Call a function with circuit breaker protection."""
if self.state == CircuitState.OPEN:
if datetime.now() - self.last_failure_time > timedelta(seconds=self.timeout_seconds):
self.state = CircuitState.HALF_OPEN
else:
raise Exception("Circuit breaker is open")
try:
result = await func(*args, **kwargs)
self.on_success()
return result
except Exception as e:
self.on_failure()
raise
def on_success(self):
"""Handle successful call."""
self.failure_count = 0
self.state = CircuitState.CLOSED
def on_failure(self):
"""Handle failed call."""
self.failure_count += 1
self.last_failure_time = datetime.now()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
```
## Best Practices
### 1. Session Management
- Always create a session before executing plans
- Use unique session IDs for tracking
- Implement session timeout mechanisms
- Clean up sessions when done
### 2. Error Handling
- Implement comprehensive error handling
- Use appropriate error codes
- Provide detailed error messages
- Log all errors for debugging
### 3. Performance
- Use connection pooling for HTTP transport
- Implement request timeouts
- Monitor response times
- Optimize message serialization
### 4. Security
- Use TLS/SSL for HTTP transport
- Implement authentication and authorization
- Validate all inputs
- Sanitize error messages
### 5. Monitoring
- Log all RPC calls
- Track execution times
- Monitor error rates
- Implement health checks
### 6. Testing
- Unit test protocol handlers
- Integration test with real agents
- Load test with multiple concurrent sessions
- Test error recovery mechanisms
## Related Documentation
- [A2A Protocol Specification](../reference/a2a_protocol_spec.md)
- [A2A Server Deployment Guide](./a2a_server_deployment.md)
- [ADR-047: ACP Standard Adoption](../adr/ADR-047-acp-standard-adoption.md)
- [ADR-048: Server Application Architecture](../adr/ADR-048-server-application-architecture.md)
+749
View File
@@ -0,0 +1,749 @@
# A2A Server Deployment Guide
## Overview
This guide provides comprehensive instructions for deploying A2A servers in production environments. It covers Docker, Kubernetes, and Helm deployments, along with security, monitoring, and operational best practices.
## Server Deployment Overview
The A2A Server is a stateless HTTP service that implements the A2A Protocol. It can be deployed in various environments:
- **Single Container**: Simple deployments with Docker
- **Kubernetes Cluster**: Scalable deployments with automatic failover
- **Helm Charts**: Declarative infrastructure as code
- **Cloud Platforms**: AWS, GCP, Azure deployments
### Architecture
```
┌─────────────────────────────────────────────────────────┐
│ Load Balancer │
│ (Nginx, HAProxy, ALB) │
└──────────────────┬──────────────────────────────────────┘
┌──────────┼──────────┐
│ │ │
┌───────▼──┐ ┌────▼────┐ ┌───▼──────┐
│ A2A Pod 1│ │ A2A Pod2│ │ A2A Pod3 │
└──────────┘ └─────────┘ └──────────┘
│ │ │
└──────────┼──────────┘
┌──────────▼──────────┐
│ Persistent Storage │
│ (Redis, PostgreSQL) │
└─────────────────────┘
```
## Docker Deployment
### Dockerfile
```dockerfile
FROM python:3.11-slim
WORKDIR /app
# Install dependencies
RUN apt-get update && apt-get install -y \
curl \
&& rm -rf /var/lib/apt/lists/*
# Copy application
COPY . .
# Install Python dependencies
RUN pip install --no-cache-dir -r requirements.txt
# Expose port
EXPOSE 8000
# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# Run application
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
```
### Docker Compose
```yaml
version: '3.8'
services:
a2a-server:
build: .
ports:
- "8000:8000"
environment:
- A2A_MODE=server
- A2A_TRANSPORT=http
- LOG_LEVEL=info
- REDIS_URL=redis://redis:6379
- DATABASE_URL=postgresql://user:password@postgres:5432/a2a
depends_on:
- redis
- postgres
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
restart: unless-stopped
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
- redis_data:/data
restart: unless-stopped
postgres:
image: postgres:15-alpine
environment:
- POSTGRES_USER=user
- POSTGRES_PASSWORD=password
- POSTGRES_DB=a2a
ports:
- "5432:5432"
volumes:
- postgres_data:/var/lib/postgresql/data
restart: unless-stopped
volumes:
redis_data:
postgres_data:
```
### Running Docker Deployment
```bash
# Build image
docker build -t cleveragents/a2a-server:latest .
# Run with Docker Compose
docker-compose up -d
# View logs
docker-compose logs -f a2a-server
# Stop services
docker-compose down
```
## Kubernetes Deployment
### Deployment Manifest
```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: a2a-server
namespace: cleveragents
labels:
app: a2a-server
version: v1
spec:
replicas: 3
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 1
maxUnavailable: 0
selector:
matchLabels:
app: a2a-server
template:
metadata:
labels:
app: a2a-server
version: v1
spec:
serviceAccountName: a2a-server
securityContext:
runAsNonRoot: true
runAsUser: 1000
fsGroup: 1000
containers:
- name: a2a-server
image: cleveragents/a2a-server:latest
imagePullPolicy: IfNotPresent
ports:
- name: http
containerPort: 8000
protocol: TCP
env:
- name: A2A_MODE
value: "server"
- name: A2A_TRANSPORT
value: "http"
- name: LOG_LEVEL
value: "info"
- name: REDIS_URL
valueFrom:
secretKeyRef:
name: a2a-secrets
key: redis-url
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: a2a-secrets
key: database-url
- name: API_KEY
valueFrom:
secretKeyRef:
name: a2a-secrets
key: api-key
resources:
requests:
cpu: 500m
memory: 512Mi
limits:
cpu: 2000m
memory: 2Gi
livenessProbe:
httpGet:
path: /health
port: http
initialDelaySeconds: 10
periodSeconds: 30
timeoutSeconds: 5
failureThreshold: 3
readinessProbe:
httpGet:
path: /health
port: http
initialDelaySeconds: 5
periodSeconds: 10
timeoutSeconds: 5
failureThreshold: 3
securityContext:
allowPrivilegeEscalation: false
readOnlyRootFilesystem: true
capabilities:
drop:
- ALL
volumeMounts:
- name: tmp
mountPath: /tmp
- name: cache
mountPath: /app/cache
volumes:
- name: tmp
emptyDir: {}
- name: cache
emptyDir: {}
affinity:
podAntiAffinity:
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
podAffinityTerm:
labelSelector:
matchExpressions:
- key: app
operator: In
values:
- a2a-server
topologyKey: kubernetes.io/hostname
```
### Service Manifest
```yaml
apiVersion: v1
kind: Service
metadata:
name: a2a-server
namespace: cleveragents
labels:
app: a2a-server
spec:
type: ClusterIP
ports:
- name: http
port: 8000
targetPort: http
protocol: TCP
selector:
app: a2a-server
```
### Ingress Manifest
```yaml
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: a2a-server
namespace: cleveragents
annotations:
cert-manager.io/cluster-issuer: letsencrypt-prod
nginx.ingress.kubernetes.io/rate-limit: "100"
nginx.ingress.kubernetes.io/ssl-redirect: "true"
spec:
ingressClassName: nginx
tls:
- hosts:
- a2a.example.com
secretName: a2a-tls
rules:
- host: a2a.example.com
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: a2a-server
port:
number: 8000
```
### Deploying to Kubernetes
```bash
# Create namespace
kubectl create namespace cleveragents
# Create secrets
kubectl create secret generic a2a-secrets \
--from-literal=redis-url=redis://redis:6379 \
--from-literal=database-url=postgresql://user:password@postgres:5432/a2a \
--from-literal=api-key=your-api-key \
-n cleveragents
# Apply manifests
kubectl apply -f deployment.yaml
kubectl apply -f service.yaml
kubectl apply -f ingress.yaml
# Check deployment status
kubectl get deployment a2a-server -n cleveragents
kubectl get pods -n cleveragents -l app=a2a-server
# View logs
kubectl logs -n cleveragents -l app=a2a-server -f
```
## Helm Configuration
### Helm Chart Structure
```
a2a-server-chart/
├── Chart.yaml
├── values.yaml
├── templates/
│ ├── deployment.yaml
│ ├── service.yaml
│ ├── ingress.yaml
│ ├── configmap.yaml
│ ├── secret.yaml
│ ├── hpa.yaml
│ └── pdb.yaml
└── README.md
```
### Chart.yaml
```yaml
apiVersion: v2
name: a2a-server
description: A Helm chart for A2A Server
type: application
version: 1.0.0
appVersion: "1.0.0"
keywords:
- a2a
- cleveragents
- agent-communication
maintainers:
- name: CleverAgents
email: support@cleveragents.com
```
### values.yaml
```yaml
replicaCount: 3
image:
repository: cleveragents/a2a-server
tag: latest
pullPolicy: IfNotPresent
service:
type: ClusterIP
port: 8000
targetPort: 8000
ingress:
enabled: true
className: nginx
annotations:
cert-manager.io/cluster-issuer: letsencrypt-prod
hosts:
- host: a2a.example.com
paths:
- path: /
pathType: Prefix
tls:
- secretName: a2a-tls
hosts:
- a2a.example.com
resources:
requests:
cpu: 500m
memory: 512Mi
limits:
cpu: 2000m
memory: 2Gi
autoscaling:
enabled: true
minReplicas: 3
maxReplicas: 10
targetCPUUtilizationPercentage: 70
targetMemoryUtilizationPercentage: 80
nodeSelector: {}
tolerations: []
affinity:
podAntiAffinity:
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
podAffinityTerm:
labelSelector:
matchExpressions:
- key: app
operator: In
values:
- a2a-server
topologyKey: kubernetes.io/hostname
config:
mode: server
transport: http
logLevel: info
sessionTimeout: 3600
maxConcurrentSessions: 100
secrets:
redisUrl: ""
databaseUrl: ""
apiKey: ""
```
### Installing Helm Chart
```bash
# Add Helm repository
helm repo add cleveragents https://charts.cleveragents.com
helm repo update
# Install chart
helm install a2a-server cleveragents/a2a-server \
--namespace cleveragents \
--create-namespace \
--values values.yaml
# Upgrade chart
helm upgrade a2a-server cleveragents/a2a-server \
--namespace cleveragents \
--values values.yaml
# Uninstall chart
helm uninstall a2a-server --namespace cleveragents
```
## Environment Variables
### Core Configuration
| Variable | Default | Description |
|----------|---------|-------------|
| `A2A_MODE` | `server` | Deployment mode (server, local) |
| `A2A_TRANSPORT` | `http` | Transport mechanism (http, stdio) |
| `A2A_HOST` | `0.0.0.0` | Server host address |
| `A2A_PORT` | `8000` | Server port |
| `LOG_LEVEL` | `info` | Logging level (debug, info, warning, error) |
### Session Configuration
| Variable | Default | Description |
|----------|---------|-------------|
| `SESSION_TIMEOUT` | `3600` | Session timeout in seconds |
| `MAX_CONCURRENT_SESSIONS` | `100` | Maximum concurrent sessions |
| `SESSION_STORAGE` | `redis` | Session storage backend |
### Database Configuration
| Variable | Default | Description |
|----------|---------|-------------|
| `DATABASE_URL` | - | Database connection URL |
| `DATABASE_POOL_SIZE` | `10` | Database connection pool size |
| `DATABASE_TIMEOUT` | `30` | Database timeout in seconds |
### Redis Configuration
| Variable | Default | Description |
|----------|---------|-------------|
| `REDIS_URL` | - | Redis connection URL |
| `REDIS_POOL_SIZE` | `10` | Redis connection pool size |
### Security Configuration
| Variable | Default | Description |
|----------|---------|-------------|
| `API_KEY` | - | API key for authentication |
| `TLS_ENABLED` | `true` | Enable TLS/SSL |
| `TLS_CERT_PATH` | - | Path to TLS certificate |
| `TLS_KEY_PATH` | - | Path to TLS private key |
## Security Configuration
### TLS/SSL Setup
```yaml
# config.yaml
server:
tls:
enabled: true
cert_path: /etc/ssl/certs/server.crt
key_path: /etc/ssl/private/server.key
min_version: "1.2"
cipher_suites:
- TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384
- TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256
```
### Authentication
```python
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthCredentials
security = HTTPBearer()
async def verify_api_key(credentials: HTTPAuthCredentials = Depends(security)):
"""Verify API key."""
if credentials.credentials != os.getenv("API_KEY"):
raise HTTPException(status_code=401, detail="Invalid API key")
return credentials.credentials
@app.post("/rpc")
async def rpc_endpoint(
request: JSONRPCRequest,
api_key: str = Depends(verify_api_key)
):
"""RPC endpoint with authentication."""
# Handle RPC call
pass
```
### RBAC Configuration
```yaml
# rbac.yaml
apiVersion: rbac.authorization.k8s.io/v1
kind: ServiceAccount
metadata:
name: a2a-server
namespace: cleveragents
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
name: a2a-server
namespace: cleveragents
rules:
- apiGroups: [""]
resources: ["configmaps"]
verbs: ["get", "list", "watch"]
- apiGroups: [""]
resources: ["secrets"]
verbs: ["get"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
name: a2a-server
namespace: cleveragents
roleRef:
apiGroup: rbac.authorization.k8s.io
kind: Role
name: a2a-server
subjects:
- kind: ServiceAccount
name: a2a-server
namespace: cleveragents
```
## Monitoring & Logging
### Prometheus Metrics
```python
from prometheus_client import Counter, Histogram, Gauge
# Define metrics
rpc_calls_total = Counter(
'a2a_rpc_calls_total',
'Total RPC calls',
['method', 'status']
)
rpc_call_duration = Histogram(
'a2a_rpc_call_duration_seconds',
'RPC call duration',
['method']
)
active_sessions = Gauge(
'a2a_active_sessions',
'Number of active sessions'
)
# Use metrics
@app.post("/rpc")
async def rpc_endpoint(request: JSONRPCRequest):
"""RPC endpoint with metrics."""
with rpc_call_duration.labels(method=request.method).time():
try:
result = await handle_rpc(request)
rpc_calls_total.labels(method=request.method, status='success').inc()
return result
except Exception as e:
rpc_calls_total.labels(method=request.method, status='error').inc()
raise
```
### Logging Configuration
```yaml
# logging.yaml
version: 1
disable_existing_loggers: false
formatters:
standard:
format: '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
json:
format: '{"timestamp": "%(asctime)s", "level": "%(levelname)s", "message": "%(message)s"}'
handlers:
console:
class: logging.StreamHandler
level: INFO
formatter: json
stream: ext://sys.stdout
file:
class: logging.handlers.RotatingFileHandler
level: DEBUG
formatter: json
filename: /var/log/a2a-server.log
maxBytes: 10485760
backupCount: 10
loggers:
a2a:
level: DEBUG
handlers: [console, file]
root:
level: INFO
handlers: [console, file]
```
### ELK Stack Integration
```yaml
apiVersion: v1
kind: ConfigMap
metadata:
name: filebeat-config
namespace: cleveragents
data:
filebeat.yml: |
filebeat.inputs:
- type: container
paths:
- '/var/lib/docker/containers/*/*.log'
processors:
- add_kubernetes_metadata:
in_cluster: true
output.elasticsearch:
hosts: ["elasticsearch:9200"]
index: "a2a-server-%{+yyyy.MM.dd}"
```
## Operational Best Practices
### Health Checks
```bash
# Check server health
curl http://localhost:8000/health
# Check readiness
curl http://localhost:8000/ready
# Check liveness
curl http://localhost:8000/live
```
### Backup and Recovery
```bash
# Backup database
pg_dump -h postgres -U user a2a > backup.sql
# Restore database
psql -h postgres -U user a2a < backup.sql
# Backup Redis
redis-cli BGSAVE
# Restore Redis
redis-cli --pipe < dump.rdb
```
### Scaling
```bash
# Horizontal scaling with Kubernetes
kubectl scale deployment a2a-server --replicas=5 -n cleveragents
# Vertical scaling
kubectl set resources deployment a2a-server \
--limits=cpu=4,memory=4Gi \
--requests=cpu=2,memory=2Gi \
-n cleveragents
```
## Related Documentation
- [A2A Protocol Specification](../reference/a2a_protocol_spec.md)
- [A2A Implementation Guide](./a2a_implementation_guide.md)
- [ADR-048: Server Application Architecture](../adr/ADR-048-server-application-architecture.md)
+871
View File
@@ -0,0 +1,871 @@
# A2A Protocol Specification
## Overview
The Agent-to-Agent (A2A) Protocol is a standardized communication protocol for inter-agent communication in the CleverAgents ecosystem. It enables agents to communicate with each other, coordinate on tasks, and manage distributed workflows.
The A2A Protocol is built on top of JSON-RPC 2.0 and extends it with CleverAgents-specific methods and semantics. It supports multiple transport mechanisms including stdio (for local/embedded mode) and HTTP (for server mode).
### Key Features
- **JSON-RPC 2.0 Compliant**: Uses the JSON-RPC 2.0 specification as the base protocol
- **Extensible**: Supports custom methods via the `_cleveragents/` namespace
- **Multi-Transport**: Works over stdio, HTTP, and other transport mechanisms
- **Bidirectional Communication**: Supports both request-response and notification patterns
- **Error Handling**: Comprehensive error codes and recovery mechanisms
- **Session Management**: Built-in session lifecycle management
- **Plan Coordination**: Support for distributed plan execution and coordination
## JSON-RPC 2.0 Foundation
The A2A Protocol is built on JSON-RPC 2.0, which defines:
### Request Format
```json
{
"jsonrpc": "2.0",
"method": "method_name",
"params": {},
"id": "request_id"
}
```
### Response Format
```json
{
"jsonrpc": "2.0",
"result": {},
"id": "request_id"
}
```
### Error Response Format
```json
{
"jsonrpc": "2.0",
"error": {
"code": -32600,
"message": "Invalid Request",
"data": {}
},
"id": "request_id"
}
```
### Notification Format (No Response Expected)
```json
{
"jsonrpc": "2.0",
"method": "method_name",
"params": {}
}
```
## CleverAgents Extension Methods
All CleverAgents-specific methods use the `_cleveragents/` namespace prefix to distinguish them from standard JSON-RPC methods.
### Session Management Methods
#### `_cleveragents/session/create`
Creates a new session for agent communication.
**Request:**
```json
{
"jsonrpc": "2.0",
"method": "_cleveragents/session/create",
"params": {
"session_id": "string",
"agent_id": "string",
"context": {}
},
"id": "request_id"
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"result": {
"session_id": "string",
"created_at": "ISO8601_timestamp",
"status": "active"
},
"id": "request_id"
}
```
#### `_cleveragents/session/close`
Closes an existing session.
**Request:**
```json
{
"jsonrpc": "2.0",
"method": "_cleveragents/session/close",
"params": {
"session_id": "string"
},
"id": "request_id"
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"result": {
"session_id": "string",
"closed_at": "ISO8601_timestamp",
"status": "closed"
},
"id": "request_id"
}
```
### Plan Management Methods
#### `_cleveragents/plan/create`
Creates a new plan for execution.
**Request:**
```json
{
"jsonrpc": "2.0",
"method": "_cleveragents/plan/create",
"params": {
"session_id": "string",
"plan_id": "string",
"plan_definition": {},
"metadata": {}
},
"id": "request_id"
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"result": {
"plan_id": "string",
"session_id": "string",
"status": "created",
"created_at": "ISO8601_timestamp"
},
"id": "request_id"
}
```
#### `_cleveragents/plan/execute`
Executes a plan.
**Request:**
```json
{
"jsonrpc": "2.0",
"method": "_cleveragents/plan/execute",
"params": {
"session_id": "string",
"plan_id": "string",
"execution_options": {}
},
"id": "request_id"
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"result": {
"plan_id": "string",
"execution_id": "string",
"status": "executing",
"started_at": "ISO8601_timestamp"
},
"id": "request_id"
}
```
#### `_cleveragents/plan/status`
Gets the status of a plan.
**Request:**
```json
{
"jsonrpc": "2.0",
"method": "_cleveragents/plan/status",
"params": {
"session_id": "string",
"plan_id": "string"
},
"id": "request_id"
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"result": {
"plan_id": "string",
"status": "executing|completed|failed|cancelled",
"progress": {
"total_steps": 10,
"completed_steps": 5,
"current_step": "step_name"
},
"result": {},
"error": null
},
"id": "request_id"
}
```
### Agent Communication Methods
#### `_cleveragents/agent/call`
Calls another agent with a request.
**Request:**
```json
{
"jsonrpc": "2.0",
"method": "_cleveragents/agent/call",
"params": {
"session_id": "string",
"target_agent_id": "string",
"method": "string",
"params": {},
"timeout": 30000
},
"id": "request_id"
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"result": {
"agent_id": "string",
"method": "string",
"result": {},
"execution_time_ms": 150
},
"id": "request_id"
}
```
#### `_cleveragents/agent/notify`
Sends a notification to another agent (no response expected).
**Request:**
```json
{
"jsonrpc": "2.0",
"method": "_cleveragents/agent/notify",
"params": {
"session_id": "string",
"target_agent_id": "string",
"event": "string",
"data": {}
}
}
```
### Skill Execution Methods
#### `_cleveragents/skill/execute`
Executes a skill.
**Request:**
```json
{
"jsonrpc": "2.0",
"method": "_cleveragents/skill/execute",
"params": {
"session_id": "string",
"skill_id": "string",
"skill_params": {},
"timeout": 30000
},
"id": "request_id"
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"result": {
"skill_id": "string",
"status": "success|failed",
"output": {},
"execution_time_ms": 250
},
"id": "request_id"
}
```
### Context Management Methods
#### `_cleveragents/context/get`
Retrieves context information.
**Request:**
```json
{
"jsonrpc": "2.0",
"method": "_cleveragents/context/get",
"params": {
"session_id": "string",
"context_key": "string"
},
"id": "request_id"
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"result": {
"context_key": "string",
"value": {},
"timestamp": "ISO8601_timestamp"
},
"id": "request_id"
}
```
#### `_cleveragents/context/set`
Sets context information.
**Request:**
```json
{
"jsonrpc": "2.0",
"method": "_cleveragents/context/set",
"params": {
"session_id": "string",
"context_key": "string",
"value": {},
"ttl": 3600
},
"id": "request_id"
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"result": {
"context_key": "string",
"status": "set",
"timestamp": "ISO8601_timestamp"
},
"id": "request_id"
}
```
## Message Types and Formats
### Message Envelope
All A2A messages follow a standard envelope format:
```json
{
"jsonrpc": "2.0",
"method": "method_name",
"params": {
"session_id": "string",
"request_id": "string",
"timestamp": "ISO8601_timestamp",
"version": "1.0"
},
"id": "unique_request_id"
}
```
### Common Parameter Types
#### Session Parameters
```json
{
"session_id": "string",
"agent_id": "string",
"context": {
"project_id": "string",
"environment": "string",
"metadata": {}
}
}
```
#### Plan Parameters
```json
{
"plan_id": "string",
"plan_definition": {
"name": "string",
"description": "string",
"steps": [
{
"id": "string",
"type": "string",
"action": "string",
"params": {}
}
]
}
}
```
#### Execution Parameters
```json
{
"execution_id": "string",
"plan_id": "string",
"status": "pending|executing|completed|failed|cancelled",
"progress": {
"total_steps": 10,
"completed_steps": 5,
"current_step": "string"
},
"result": {},
"error": {
"code": "string",
"message": "string",
"details": {}
}
}
```
## Transport Details
### Stdio Transport (Local Mode)
The stdio transport is used for local agent communication, typically for embedded or single-process scenarios.
**Characteristics:**
- Uses standard input/output streams
- Synchronous request-response pattern
- No network overhead
- Suitable for local testing and development
**Message Format:**
```
<message_length>\n
<json_message>
```
**Example:**
```
145
{"jsonrpc":"2.0","method":"_cleveragents/session/create","params":{"session_id":"sess_123"},"id":"req_1"}
```
### HTTP Transport (Server Mode)
The HTTP transport is used for remote agent communication over a network.
**Characteristics:**
- Uses HTTP/1.1 or HTTP/2
- RESTful endpoints
- Asynchronous support via webhooks
- Suitable for distributed systems
**Endpoints:**
#### POST /rpc
Main RPC endpoint for method calls.
**Request:**
```http
POST /rpc HTTP/1.1
Content-Type: application/json
```
**Response:**
```http
HTTP/1.1 200 OK
Content-Type: application/json
```
#### POST /events
Event notification endpoint.
**Request:**
```http
POST /events HTTP/1.1
Content-Type: application/json
```
#### GET /health
Health check endpoint.
**Response:**
```http
HTTP/1.1 200 OK
Content-Type: application/json
```
### Transport Selection
The transport mechanism is selected based on the deployment mode:
- **Local Mode**: Uses stdio transport for direct process communication
- **Server Mode**: Uses HTTP transport for network communication
- **Hybrid Mode**: Can use both transports simultaneously
## Error Handling
### Standard JSON-RPC Error Codes
| Code | Message | Description |
|------|---------|-------------|
| -32700 | Parse error | Invalid JSON was received |
| -32600 | Invalid Request | The JSON sent is not a valid Request object |
| -32601 | Method not found | The method does not exist or is not available |
| -32602 | Invalid params | Invalid method parameter(s) |
| -32603 | Internal error | Internal JSON-RPC error |
| -32000 to -32099 | Server error | Reserved for implementation-defined server errors |
### CleverAgents-Specific Error Codes
| Code | Message | Description |
|------|---------|-------------|
| -32100 | Session not found | The specified session does not exist |
| -32101 | Plan not found | The specified plan does not exist |
| -32102 | Agent not found | The specified agent does not exist |
| -32103 | Skill execution failed | Skill execution failed with error |
| -32104 | Timeout | Request timed out |
| -32105 | Authorization failed | User is not authorized for this operation |
| -32106 | Resource exhausted | System resource limit exceeded |
| -32107 | Invalid state | Operation not valid in current state |
### Error Response Example
```json
{
"jsonrpc": "2.0",
"error": {
"code": -32100,
"message": "Session not found",
"data": {
"session_id": "sess_123",
"timestamp": "2024-04-19T10:30:00Z",
"request_id": "req_1"
}
},
"id": "req_1"
}
```
### Error Recovery
The A2A Protocol supports several error recovery mechanisms:
1. **Automatic Retry**: Failed requests can be automatically retried with exponential backoff
2. **Circuit Breaker**: Failing agents can be temporarily isolated to prevent cascading failures
3. **Fallback**: Alternative agents or skills can be used if primary ones fail
4. **Checkpoint/Restore**: Plan execution can be checkpointed and restored from failure points
## Protocol Examples
### Example 1: Creating and Executing a Plan
```json
// Step 1: Create a session
{
"jsonrpc": "2.0",
"method": "_cleveragents/session/create",
"params": {
"session_id": "sess_abc123",
"agent_id": "agent_main",
"context": {
"project_id": "proj_123",
"environment": "production"
}
},
"id": "req_1"
}
// Response
{
"jsonrpc": "2.0",
"result": {
"session_id": "sess_abc123",
"created_at": "2024-04-19T10:30:00Z",
"status": "active"
},
"id": "req_1"
}
// Step 2: Create a plan
{
"jsonrpc": "2.0",
"method": "_cleveragents/plan/create",
"params": {
"session_id": "sess_abc123",
"plan_id": "plan_xyz789",
"plan_definition": {
"name": "Deploy Application",
"steps": [
{
"id": "step_1",
"type": "skill",
"action": "build_application",
"params": {"version": "1.0"}
},
{
"id": "step_2",
"type": "skill",
"action": "deploy_application",
"params": {"environment": "production"}
}
]
}
},
"id": "req_2"
}
// Response
{
"jsonrpc": "2.0",
"result": {
"plan_id": "plan_xyz789",
"session_id": "sess_abc123",
"status": "created",
"created_at": "2024-04-19T10:30:05Z"
},
"id": "req_2"
}
// Step 3: Execute the plan
{
"jsonrpc": "2.0",
"method": "_cleveragents/plan/execute",
"params": {
"session_id": "sess_abc123",
"plan_id": "plan_xyz789",
"execution_options": {
"parallel": false,
"timeout": 3600000
}
},
"id": "req_3"
}
// Response
{
"jsonrpc": "2.0",
"result": {
"plan_id": "plan_xyz789",
"execution_id": "exec_123",
"status": "executing",
"started_at": "2024-04-19T10:30:10Z"
},
"id": "req_3"
}
// Step 4: Check plan status
{
"jsonrpc": "2.0",
"method": "_cleveragents/plan/status",
"params": {
"session_id": "sess_abc123",
"plan_id": "plan_xyz789"
},
"id": "req_4"
}
// Response
{
"jsonrpc": "2.0",
"result": {
"plan_id": "plan_xyz789",
"status": "completed",
"progress": {
"total_steps": 2,
"completed_steps": 2,
"current_step": null
},
"result": {
"deployment_id": "deploy_456",
"status": "success"
},
"error": null
},
"id": "req_4"
}
```
### Example 2: Agent-to-Agent Communication
```json
// Agent A calls Agent B
{
"jsonrpc": "2.0",
"method": "_cleveragents/agent/call",
"params": {
"session_id": "sess_abc123",
"target_agent_id": "agent_worker",
"method": "process_data",
"params": {
"data": [1, 2, 3, 4, 5],
"operation": "sum"
},
"timeout": 5000
},
"id": "req_5"
}
// Response from Agent B
{
"jsonrpc": "2.0",
"result": {
"agent_id": "agent_worker",
"method": "process_data",
"result": {
"sum": 15,
"count": 5,
"average": 3
},
"execution_time_ms": 125
},
"id": "req_5"
}
```
### Example 3: Skill Execution
```json
// Execute a skill
{
"jsonrpc": "2.0",
"method": "_cleveragents/skill/execute",
"params": {
"session_id": "sess_abc123",
"skill_id": "skill_email_sender",
"skill_params": {
"recipient": "user@example.com",
"subject": "Hello",
"body": "This is a test email"
},
"timeout": 10000
},
"id": "req_6"
}
// Response
{
"jsonrpc": "2.0",
"result": {
"skill_id": "skill_email_sender",
"status": "success",
"output": {
"message_id": "msg_789",
"sent_at": "2024-04-19T10:30:30Z"
},
"execution_time_ms": 450
},
"id": "req_6"
}
```
## Protocol Versioning
The A2A Protocol uses semantic versioning:
- **Major Version**: Breaking changes to the protocol
- **Minor Version**: Backward-compatible additions
- **Patch Version**: Bug fixes and non-breaking changes
Current version: **1.0.0**
## Security Considerations
### Authentication
All A2A communications should be authenticated using:
- API keys for HTTP transport
- Session tokens for long-lived connections
- TLS/SSL for encrypted communication
### Authorization
Authorization is enforced at multiple levels:
- Session-level: Who can access a session
- Plan-level: Who can execute a plan
- Skill-level: Who can execute a skill
- Agent-level: Who can call an agent
### Message Integrity
All messages should include:
- Request IDs for tracking
- Timestamps for audit trails
- Digital signatures for critical operations
## Related Documentation
- [A2A Implementation Guide](../guides/a2a_implementation_guide.md)
- [A2A Server Deployment Guide](../guides/a2a_server_deployment.md)
- [ADR-047: ACP Standard Adoption](../adr/ADR-047-acp-standard-adoption.md)
- [ADR-048: Server Application Architecture](../adr/ADR-048-server-application-architecture.md)
- [ADR-026: Agent Client Protocol](../adr/ADR-026-agent-client-protocol.md)
- TLS/SSL for encrypted communication
### Authorization
Authorization is enforced at multiple levels:
- Session-level: Who can access a session
- Plan-level: Who can execute a plan
- Skill-level: Who can execute a skill
- Agent-level: Who can call an agent
### Message Integrity
All messages should include:
- Request IDs for tracking
- Timestamps for audit trails
- Digital signatures for critical operations
## Related Documentation
- [A2A Implementation Guide](../guides/a2a_implementation_guide.md)
- [A2A Server Deployment Guide](../guides/a2a_server_deployment.md)
- [ADR-047: ACP Standard Adoption](../adr/ADR-047-acp-standard-adoption.md)
- [ADR-048: Server Application Architecture](../adr/ADR-048-server-application-architecture.md)
- [ADR-026: Agent Client Protocol](../adr/ADR-026-agent-client-protocol.md)