Back to Documentation
API Reference
Complete API documentation for AMP SDKs and protocol implementations
Python SDK
The Python SDK provides async/await support, type hints, and framework integrations.
AMPAgent
Base class for creating AMP-compatible agents.
class AMPAgent:
def __init__(
self,
agent_id: str,
name: str,
description: str = "",
version: str = "1.0.0",
framework: str = "custom"
)
async def connect(self, endpoint: str, **kwargs) -> None
async def disconnect(self) -> None
async def run(self) -> None
def register_capability(self, capability: Capability) -> None
def unregister_capability(self, capability_id: str) -> None
async def invoke_capability(
self,
capability: str,
payload: Dict[str, Any],
target_agent: Optional[str] = None,
timeout: int = 30000
) -> Dict[str, Any]
async def broadcast_event(
self,
event_type: str,
payload: Dict[str, Any]
) -> NoneParameters
- •
agent_id- Unique identifier - •
name- Human-readable name - •
description- Agent description - •
version- Agent version - •
framework- Source framework
Methods
- •
connect()- Connect to mesh - •
disconnect()- Disconnect - •
run()- Start event loop - •
register_capability()- Add capability - •
invoke_capability()- Call capability
Capability
Represents a capability that an agent can provide.
class Capability:
def __init__(
self,
id: str,
version: str,
handler: Callable,
description: str = "",
input_schema: Optional[Dict] = None,
output_schema: Optional[Dict] = None,
constraints: Optional[Dict] = None
)
async def invoke(self, request: AMPRequest) -> Any
def validate_input(self, data: Any) -> bool
def validate_output(self, data: Any) -> boolExample Usage
capability = Capability(
id="text-analysis",
version="1.0.0",
handler=analyze_text,
description="Analyzes text for sentiment and entities",
input_schema={
"type": "object",
"properties": {
"text": {"type": "string", "maxLength": 1000}
},
"required": ["text"]
},
output_schema={
"type": "object",
"properties": {
"sentiment": {"type": "string"},
"entities": {"type": "array"}
}
},
constraints={
"max_response_time_ms": 5000,
"rate_limit": "100/minute"
}
)AMPClient
Client for connecting to and interacting with an agent mesh.
class AMPClient:
def __init__(
self,
client_id: Optional[str] = None,
auth_token: Optional[str] = None
)
async def connect(self, endpoint: str, **kwargs) -> None
async def disconnect(self) -> None
async def discover_agents(
self,
capability: Optional[str] = None,
framework: Optional[str] = None
) -> List[AgentInfo]
async def invoke_capability(
self,
capability: str,
payload: Dict[str, Any],
target_agent: Optional[str] = None,
timeout: int = 30000
) -> Dict[str, Any]
async def subscribe_events(
self,
event_types: List[str],
handler: Callable
) -> str # Returns subscription IDTypeScript SDK
Full TypeScript support with React hooks and browser compatibility.
AMPAgent Interface
interface AMPAgentConfig {
agentId: string
name: string
description?: string
version?: string
framework?: string
}
class AMPAgent {
constructor(config: AMPAgentConfig)
async connect(endpoint: string, options?: ConnectionOptions): Promise<void>
async disconnect(): Promise<void>
async run(): Promise<void>
registerCapability(capability: Capability): void
unregisterCapability(capabilityId: string): void
async invokeCapability<T = any>(
capability: string,
payload: any,
options?: InvokeOptions
): Promise<T>
async broadcastEvent(eventType: string, payload: any): Promise<void>
on(event: string, handler: EventHandler): void
off(event: string, handler: EventHandler): void
}React Hooks
Custom hooks for React applications.
// Connect to agent mesh
const { agent, isConnected, error } = useAMPAgent({
agentId: 'react-agent-001',
name: 'React Agent',
endpoint: 'ws://localhost:8080/amp'
})
// Invoke capabilities
const { invoke, loading, result, error } = useCapability(agent)
// Subscribe to events
useAMPEvents(agent, ['agent_registered', 'capability_updated'], (event) => {
console.log('Received event:', event)
})
// Discover agents
const { agents, refresh } = useAgentDiscovery(agent, {
capability: 'text-analysis'
})Example React Component
import { useAMPAgent, useCapability } from '@agentmeshprotocol/react'
function SentimentAnalyzer() {
const { agent, isConnected } = useAMPAgent({
agentId: 'sentiment-ui',
endpoint: 'ws://localhost:8080/amp'
})
const { invoke, loading, result } = useCapability(agent)
const [text, setText] = useState('')
const analyzeSentiment = async () => {
const response = await invoke('sentiment-analysis', { text })
console.log('Sentiment:', response.sentiment)
}
return (
<div>
<textarea value={text} onChange={(e) => setText(e.target.value)} />
<button onClick={analyzeSentiment} disabled={!isConnected || loading}>
{loading ? 'Analyzing...' : 'Analyze Sentiment'}
</button>
{result && <div>Result: {JSON.stringify(result)}</div>}
</div>
)
}HTTP REST API
Direct HTTP endpoints for language-agnostic integration.
POST /amp/v1/invoke
Invoke a capability on an agent in the mesh.
Request
POST /amp/v1/invoke
Content-Type: application/json
Authorization: Bearer <token>
{
"protocol": "AMP/1.0",
"message": {
"id": "req-123",
"type": "request",
"timestamp": "2025-01-27T10:30:00Z",
"source": {
"agent_id": "client-001"
},
"destination": {
"capability": "sentiment-analysis"
},
"payload": {
"text": "Hello world!"
}
}
}Response
HTTP/1.1 200 OK
Content-Type: application/json
{
"protocol": "AMP/1.0",
"message": {
"id": "resp-123",
"type": "response",
"timestamp": "2025-01-27T10:30:01Z",
"headers": {
"correlation_id": "req-123"
},
"payload": {
"sentiment": "positive",
"confidence": 0.8
}
}
}GET /amp/v1/agents
Discover agents and their capabilities.
GET /amp/v1/agents?capability=text-analysis&framework=langchain
{
"agents": [
{
"agent_id": "nlp-001",
"name": "NLP Processor",
"framework": "langchain",
"version": "1.0.0",
"capabilities": [
{
"id": "text-analysis",
"version": "1.2.0",
"description": "Advanced text analysis",
"constraints": {
"max_response_time_ms": 5000,
"rate_limit": "100/minute"
}
}
],
"status": "online",
"last_seen": "2025-01-27T10:29:55Z"
}
]
}WebSocket /amp/v1/ws
Real-time bidirectional communication.
// Connect to WebSocket
const ws = new WebSocket('ws://localhost:8080/amp/v1/ws')
// Send AMP message
ws.send(JSON.stringify({
protocol: "AMP/1.0",
message: {
id: "msg-001",
type: "request",
destination: { capability: "text-analysis" },
payload: { text: "Hello AMP!" }
}
}))
// Receive responses and events
ws.onmessage = (event) => {
const ampMessage = JSON.parse(event.data)
console.log('Received:', ampMessage)
}Error Handling
Standard error codes and handling patterns across all AMP implementations.
CAPABILITY_NOT_FOUND
The requested capability is not available in the mesh.
{
"error_code": "CAPABILITY_NOT_FOUND",
"error_message": "Capability 'image-generation' not found",
"suggested_alternatives": ["text-generation", "image-analysis"]
}VALIDATION_ERROR
Input or output data doesn't match the capability schema.
{
"error_code": "VALIDATION_ERROR",
"error_message": "Input validation failed",
"details": {
"field": "text",
"expected": "string",
"received": "number"
}
}TIMEOUT
The capability invocation exceeded the timeout limit.
{
"error_code": "TIMEOUT",
"error_message": "Capability invocation timed out after 30000ms",
"retry_after": 5000
}RATE_LIMITED
The request was rejected due to rate limiting.
{
"error_code": "RATE_LIMITED",
"error_message": "Rate limit exceeded: 100 requests per minute",
"retry_after": 15000,
"current_limit": "100/minute"
}