Overview
Simple, runnable examples for each asyncmcp transport type.SQS Example
Server
#!/usr/bin/env python3
import asyncio
import boto3
from mcp.server.lowlevel import Server
from mcp import types
from asyncmcp.sqs import sqs_server
from asyncmcp import SqsServerConfig
# Create MCP server
app = Server("hello-world")
@app.list_tools()
async def list_tools():
return [
types.Tool(
name="greet",
description="Say hello",
inputSchema={
"type": "object",
"properties": {
"name": {"type": "string"}
},
"required": ["name"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "greet":
return [types.TextContent(
type="text",
text=f"Hello, {arguments['name']}!"
)]
async def main():
config = SqsServerConfig(
read_queue_url="http://localhost:4566/000000000000/mcp-requests"
)
sqs_client = boto3.client(
'sqs',
endpoint_url='http://localhost:4566',
region_name='us-east-1',
aws_access_key_id='test',
aws_secret_access_key='test'
)
async with sqs_server(config, sqs_client) as (read, write):
await app.run(read, write, types.InitializationOptions())
if __name__ == "__main__":
asyncio.run(main())
Client
#!/usr/bin/env python3
import asyncio
import boto3
from mcp.client import ClientSession
from asyncmcp.sqs import sqs_client
from asyncmcp import SqsClientConfig
async def main():
config = SqsClientConfig(
read_queue_url="http://localhost:4566/000000000000/mcp-requests",
response_queue_url="http://localhost:4566/000000000000/mcp-responses"
)
sqs_boto_client = boto3.client(
'sqs',
endpoint_url='http://localhost:4566',
region_name='us-east-1',
aws_access_key_id='test',
aws_secret_access_key='test'
)
async with sqs_client(config, sqs_boto_client) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
result = await session.call_tool(
"greet",
arguments={"name": "asyncmcp"}
)
print(result.content[0].text)
# Output: Hello, asyncmcp!
if __name__ == "__main__":
asyncio.run(main())
Webhook Example
Server
#!/usr/bin/env python3
from fastapi import FastAPI
import uvicorn
from mcp.server.lowlevel import Server
from mcp import types
from asyncmcp.webhook.manager import WebhookSessionManager
from asyncmcp import WebhookServerConfig
app = FastAPI()
mcp_server = Server("webhook-example")
@mcp_server.list_tools()
async def list_tools():
return [
types.Tool(
name="echo",
description="Echo back the input",
inputSchema={
"type": "object",
"properties": {
"message": {"type": "string"}
},
"required": ["message"]
}
)
]
@mcp_server.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "echo":
return [types.TextContent(
type="text",
text=f"Echo: {arguments['message']}"
)]
config = WebhookServerConfig()
session_manager = WebhookSessionManager(mcp_server, config)
app.mount("/mcp", session_manager.asgi_app())
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
Client
#!/usr/bin/env python3
import asyncio
from fastapi import FastAPI, Request
import uvicorn
from mcp.client import ClientSession
from asyncmcp import webhook_client, WebhookClientConfig
webhook_app = FastAPI()
@webhook_app.post("/webhook")
async def receive_webhook(request: Request):
# Handle webhook response
body = await request.json()
print(f"Received: {body}")
return {"status": "ok"}
async def run_client():
config = WebhookClientConfig(
server_url="http://localhost:8000/mcp/request",
webhook_url="http://localhost:8001/webhook"
)
async with webhook_client(config) as (read, write, client):
async with ClientSession(read, write) as session:
await session.initialize()
result = await session.call_tool(
"echo",
arguments={"message": "Hello Webhooks!"}
)
print(result.content[0].text)
# Output: Echo: Hello Webhooks!
async def main():
# Start webhook server
webhook_config = uvicorn.Config(
webhook_app, host="0.0.0.0", port=8001, log_level="error"
)
webhook_server = uvicorn.Server(webhook_config)
webhook_task = asyncio.create_task(webhook_server.serve())
await asyncio.sleep(1)
# Run client
await run_client()
webhook_task.cancel()
if __name__ == "__main__":
asyncio.run(main())
Proxy Example
#!/usr/bin/env python3
import asyncio
import boto3
from asyncmcp.proxy import create_proxy_server
from asyncmcp.sqs.utils import SqsClientConfig
async def main():
# Configure SQS backend
backend_config = SqsClientConfig(
read_queue_url="http://localhost:4566/000000000000/mcp-requests",
response_queue_url="http://localhost:4566/000000000000/mcp-responses"
)
sqs_client = boto3.client(
'sqs',
endpoint_url='http://localhost:4566',
region_name='us-east-1',
aws_access_key_id='test',
aws_secret_access_key='test'
)
# Create proxy server
proxy = create_proxy_server(
backend_transport="sqs",
backend_config=backend_config,
backend_clients={"sqs_client": sqs_client},
port=8080
)
print("Proxy running on http://localhost:8080/mcp")
await proxy.run()
if __name__ == "__main__":
asyncio.run(main())
Running the Examples
1
Start LocalStack
localstack start
2
Setup Resources
uv run examples/setup.py
3
Run Server
uv run server.py
4
Run Client
uv run client.py
Next Steps
SQS Transport
Learn about SQS transport
Local Development
Set up development environment