Documentation Index Fetch the complete documentation index at: https://asyncmcp.mintlify.app/llms.txt
Use this file to discover all available pages before exploring further.
Overview
Set up a complete local development environment for asyncmcp using LocalStack to simulate AWS services.
Prerequisites
Install LocalStack
# Using pip
pip install localstack
# Using Docker
docker pull localstack/localstack
# Using Homebrew (macOS)
brew install localstack
Install AWS CLI
# Configure for LocalStack
aws configure set aws_access_key_id test
aws configure set aws_secret_access_key test
aws configure set region us-east-1
Install asyncmcp
# Clone repository
git clone https://github.com/bh-rat/asyncmcp.git
cd asyncmcp
# Install with uv
uv sync
Starting LocalStack
CLI
Docker
Docker Compose
# Start LocalStack
localstack start
# Verify it's running
localstack status
docker run -d \
--name localstack \
-p 4566:4566 \
-e SERVICES=sqs,sns \
-e DEBUG= 1 \
localstack/localstack
# docker-compose.yml
version : '3.8'
services :
localstack :
image : localstack/localstack
ports :
- "4566:4566"
environment :
- SERVICES=sqs,sns
- DEBUG=1
- DATA_DIR=/tmp/localstack/data
volumes :
- ./localstack:/tmp/localstack
Setting Up Resources
Create SQS Queues
#!/usr/bin/env python3
import boto3
# LocalStack configuration
config = {
"endpoint_url" : "http://localhost:4566" ,
"region_name" : "us-east-1" ,
"aws_access_key_id" : "test" ,
"aws_secret_access_key" : "test"
}
sqs = boto3.client( "sqs" , ** config)
# Create queues
queues = [
"mcp-requests" ,
"mcp-responses" ,
"mcp-processor" ,
"mcp-consumer"
]
for queue_name in queues:
response = sqs.create_queue(
QueueName = queue_name,
Attributes = {
'ReceiveMessageWaitTimeSeconds' : '20' ,
'VisibilityTimeout' : '30'
}
)
print ( f "✅ Created queue: { response[ 'QueueUrl' ] } " )
Create SNS Topics
sns = boto3.client( "sns" , ** config)
# Create topics
topics = [
"mcp-requests" ,
"mcp-responses"
]
for topic_name in topics:
response = sns.create_topic( Name = topic_name)
print ( f "✅ Created topic: { response[ 'TopicArn' ] } " )
# Subscribe queues to topics
if topic_name == "mcp-requests" :
sns.subscribe(
TopicArn = response[ 'TopicArn' ],
Protocol = 'sqs' ,
Endpoint = f "arn:aws:sqs:us-east-1:000000000000:mcp-processor"
)
Running Examples
Basic SQS Example
# Terminal 1: Start server
uv run examples/website_server.py --transport sqs
# Terminal 2: Start client
uv run examples/website_client.py --transport sqs
SNS+SQS Example
# Terminal 1: Start server
uv run examples/website_server.py
# Terminal 2: Start client
uv run examples/website_client.py
Webhook Example
# Terminal 1: Start server
uv run examples/webhook_server.py --server-port 8000
# Terminal 2: Start client
uv run examples/webhook_client.py --server-port 8000 --webhook-port 8001
Development Workflow
Start Development
# Run your MCP server/client
uv run your_server.py
Debugging Tips
# List SQS queues
aws --endpoint-url=http://localhost:4566 sqs list-queues
# List SNS topics
aws --endpoint-url=http://localhost:4566 sns list-topics
# Check queue depth
aws --endpoint-url=http://localhost:4566 sqs get-queue-attributes \
--queue-url http://localhost:4566/000000000000/mcp-requests \
--attribute-names ApproximateNumberOfMessages
Environment Variables
# .env file for development
AWS_ENDPOINT_URL = http://localhost:4566
AWS_REGION = us-east-1
AWS_ACCESS_KEY_ID = test
AWS_SECRET_ACCESS_KEY = test
LOCALSTACK_HOST = localhost:4566
Running Tests
# Run all tests
uv run pytest
# Run with coverage
uv run pytest --cov=asyncmcp --cov-report=html
# Run specific markers
uv run pytest -m unit
uv run pytest -m integration
# Run specific test module
uv run pytest tests/sqs/test_integration.py
# Run tests with specific markers
uv run pytest -m integration
uv run pytest -m unit
Next Steps
Quickstart Run your first example