Agent Mesh Protocol

What is AMP?

The Agent Mesh Protocol (AMP) is an open standard that enables AI agents built with different frameworks to communicate and collaborate seamlessly. Whether you're using LangChain, CrewAI, AutoGen, or building custom agents, AMP provides the interoperability layer you need.

Key Benefits

Framework-agnostic communication
Automatic capability discovery
Built-in security and authentication

Scalable mesh architecture
Production-ready implementations
Open source and extensible

Installation

Choose the SDK that matches your development environment:

🐍Python SDK

pip install agentmeshprotocol

Supports Python 3.8+ with async/await, type hints, and framework integrations.

View Python SDK

🟨TypeScript SDK

npm install @agentmeshprotocol/sdk

Works in Node.js and browsers with full TypeScript support and React hooks.

View TypeScript SDK

Your First AMP Agent

Let's create a simple agent that can analyze text sentiment and respond to requests.

Python Implementation

from amp import AMPAgent, Capability
import asyncio

class SentimentAgent(AMPAgent):
    def __init__(self):
        super().__init__(
            agent_id="sentiment-analyzer-001",
            name="Sentiment Analysis Agent",
            description="Analyzes text sentiment using ML models",
            version="1.0.0"
        )
        
        # Register capabilities
        self.register_capability(
            Capability(
                id="sentiment-analysis",
                version="1.0.0",
                handler=self.analyze_sentiment,
                input_schema={
                    "type": "object",
                    "properties": {
                        "text": {"type": "string", "maxLength": 1000}
                    },
                    "required": ["text"]
                },
                output_schema={
                    "type": "object",
                    "properties": {
                        "sentiment": {"type": "string"},
                        "confidence": {"type": "number"}
                    }
                }
            )
        )
    
    async def analyze_sentiment(self, request):
        text = request.payload.get("text", "")
        
        # Simple sentiment analysis (replace with your ML model)
        positive_words = ["good", "great", "excellent", "amazing", "love"]
        negative_words = ["bad", "terrible", "awful", "hate", "horrible"]
        
        text_lower = text.lower()
        pos_count = sum(1 for word in positive_words if word in text_lower)
        neg_count = sum(1 for word in negative_words if word in text_lower)
        
        if pos_count > neg_count:
            sentiment = "positive"
            confidence = min(0.6 + (pos_count - neg_count) * 0.1, 0.95)
        elif neg_count > pos_count:
            sentiment = "negative"
            confidence = min(0.6 + (neg_count - pos_count) * 0.1, 0.95)
        else:
            sentiment = "neutral"
            confidence = 0.5
        
        return {
            "sentiment": sentiment,
            "confidence": confidence,
            "analysis_time": "2025-01-27T10:30:00Z"
        }

async def main():
    agent = SentimentAgent()
    
    # Connect to AMP mesh (replace with your mesh endpoint)
    await agent.connect("ws://localhost:8080/amp")
    
    print("✅ Sentiment agent connected and ready!")
    
    # Keep the agent running
    try:
        await agent.run()
    except KeyboardInterrupt:
        print("🛑 Shutting down agent...")
        await agent.disconnect()

if __name__ == "__main__":
    asyncio.run(main())

TypeScript Implementation

import { AMPAgent, Capability, AMPRequest, AMPResponse } from '@agentmeshprotocol/sdk'

interface SentimentRequest {
  text: string
}

interface SentimentResponse {
  sentiment: 'positive' | 'negative' | 'neutral'
  confidence: number
  analysis_time: string
}

class SentimentAgent extends AMPAgent {
  constructor() {
    super({
      agentId: 'sentiment-analyzer-001',
      name: 'Sentiment Analysis Agent',
      description: 'Analyzes text sentiment using ML models',
      version: '1.0.0'
    })

    // Register sentiment analysis capability
    this.registerCapability(new Capability({
      id: 'sentiment-analysis',
      version: '1.0.0',
      handler: this.analyzeSentiment.bind(this),
      inputSchema: {
        type: 'object',
        properties: {
          text: { type: 'string', maxLength: 1000 }
        },
        required: ['text']
      },
      outputSchema: {
        type: 'object',
        properties: {
          sentiment: { type: 'string' },
          confidence: { type: 'number' }
        }
      }
    }))
  }

  private async analyzeSentiment(
    request: AMPRequest<SentimentRequest>
  ): Promise<SentimentResponse> {
    const { text } = request.payload

    // Simple sentiment analysis
    const positiveWords = ['good', 'great', 'excellent', 'amazing', 'love']
    const negativeWords = ['bad', 'terrible', 'awful', 'hate', 'horrible']

    const textLower = text.toLowerCase()
    const posCount = positiveWords.filter(word => textLower.includes(word)).length
    const negCount = negativeWords.filter(word => textLower.includes(word)).length

    let sentiment: 'positive' | 'negative' | 'neutral'
    let confidence: number

    if (posCount > negCount) {
      sentiment = 'positive'
      confidence = Math.min(0.6 + (posCount - negCount) * 0.1, 0.95)
    } else if (negCount > posCount) {
      sentiment = 'negative'
      confidence = Math.min(0.6 + (negCount - posCount) * 0.1, 0.95)
    } else {
      sentiment = 'neutral'
      confidence = 0.5
    }

    return {
      sentiment,
      confidence,
      analysis_time: new Date().toISOString()
    }
  }
}

async function main() {
  const agent = new SentimentAgent()

  // Connect to AMP mesh
  await agent.connect('ws://localhost:8080/amp')
  
  console.log('✅ Sentiment agent connected and ready!')

  // Handle graceful shutdown
  process.on('SIGINT', async () => {
    console.log('🛑 Shutting down agent...')
    await agent.disconnect()
    process.exit(0)
  })

  // Keep the agent running
  await agent.run()
}

main().catch(console.error)

Using Your Agent

Once your agent is running, other agents can discover and use its capabilities.

Client Example

# Python client
from amp import AMPClient

async def test_sentiment_agent():
    client = AMPClient()
    await client.connect("ws://localhost:8080/amp")
    
    # Invoke sentiment analysis
    response = await client.invoke_capability(
        capability="sentiment-analysis",
        payload={"text": "I love using AMP for agent communication!"}
    )
    
    print(f"Sentiment: {response['sentiment']}")
    print(f"Confidence: {response['confidence']}")
    
    await client.disconnect()

Expected Response

{
  "protocol": "AMP/1.0",
  "message": {
    "type": "response",
    "payload": {
      "sentiment": "positive",
      "confidence": 0.7,
      "analysis_time": "2025-01-27T10:30:00Z"
    },
    "status": "success"
  }
}

Framework Integration

AMP provides adapters for popular AI frameworks, making integration seamless.

🦜 LangChain Integration

from langchain.agents import AgentExecutor
from amp.integrations.langchain import LangChainAMPAdapter

# Wrap your existing LangChain agent
executor = AgentExecutor.from_agent_and_tools(agent, tools)
amp_agent = LangChainAMPAdapter(
    executor=executor,
    agent_id="langchain-agent-001",
    capabilities=["question-answering", "tool-use"]
)

await amp_agent.connect("ws://localhost:8080/amp")

👥 CrewAI Integration

from crewai import Crew, Agent, Task
from amp.integrations.crewai import CrewAIAMPAdapter

# Your existing CrewAI setup
crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])

# Make it AMP-compatible
amp_crew = CrewAIAMPAdapter(
    crew=crew,
    agent_id="content-crew-001",
    capabilities=["research", "writing", "content-creation"]
)

await amp_crew.connect("ws://localhost:8080/amp")

🔧 AutoGen Integration

from autogen import ConversableAgent
from amp.integrations.autogen import AutoGenAMPAdapter

# Wrap AutoGen agents
autogen_agent = ConversableAgent(name="assistant", llm_config=config)
amp_agent = AutoGenAMPAdapter(
    agent=autogen_agent,
    agent_id="autogen-assistant-001",
    capabilities=["conversation", "code-generation"]
)

await amp_agent.connect("ws://localhost:8080/amp")

Next Steps

Learn More

Community

Troubleshooting

🔌 Connection Issues

Ensure your mesh endpoint is accessible and supports WebSocket connections. Check firewall settings and network connectivity.

📋 Schema Validation Errors

Verify that your capability input/output schemas match the actual data being sent. Use the validation tools provided in the SDK.

🔍 Agent Discovery Problems

Make sure agents are properly registered with unique IDs and that capability names follow the standard taxonomy.

Getting Started