# asyncmcp - Async transport layers for MCP
[](LICENSE)
[](https://www.python.org/downloads/)
---
## Overview
A regular MCP Server but working over queues :
https://github.com/user-attachments/assets/4b775ff8-02ae-4730-a822-3e1cedf9d744
Quoting from the [official description](https://modelcontextprotocol.io/introduction) :<br/>
> MCP is an open protocol that standardizes how applications provide context to LLMs.
But a lot of this context is not always readily available and takes time for the applications to process - think batch processing APIs, webhooks or queues. In these cases with the current transport layers, the MCP server would have to expose a light-weight polling wrapper in the MCP layer to allow waiting and polling for the tasks to be done. Although SSE does provide async functionalities but it comes with caveats. <br/>
asyncmcp explores supporting more of the async transport layer implementations for MCP clients and servers, beyond the officially supported stdio and Streamable Http transports.
The whole idea of an **MCP server with async transport layer** is that it doesn't have to respond immediately to any requests. It can choose to direct them to internal queues for processing and the client doesn't have to stick around for the response.
## Current capabilities
### Transport layer : sns-sqs
- Server : Transport layer that listens to a queue for MCP requests and writes the responses to a topic
- Client : Transport layer that writes requests to a topic and listens to a queue for responses
## Installation and Usage
```bash
# Using uv (recommended)
uv add asyncmcp
```
```bash
# Using pip
pip install asyncmcp
```
### Basic server setup
Note : we don't support FastMCP yet. The examples in this repo uses the [basic way of creating MCP servers and client](https://modelcontextprotocol.io/docs/concepts/architecture#implementation-example)
<br/>
Here's a basic example of implementing an MCP server which supports sns-sqs as the transport layer:
```python
import boto3
from asyncmcp.sns_sqs.server import sns_sqs_server
from asyncmcp import SnsSqsTransportConfig
# Configure transport
config = SnsSqsTransportConfig(
sqs_queue_url="https://sqs.region.amazonaws.com/account/service-queue",
sns_topic_arn="arn:aws:sns:region:account:mcp-responses"
) # more configurable params available.
# Create AWS clients
sqs_client = boto3.client('sqs')
sns_client = boto3.client('sns')
async def main():
async with sns_sqs_server(config, sqs_client, sns_client) as (read_stream, write_stream):
# Your MCP server logic here
pass
```
### Basic client setup
Here's a basic example of implementing an MCP client which supports sns-sqs as the transport layer:
```python
import boto3
from asyncmcp.sns_sqs.client import sns_sqs_client
from asyncmcp import SnsSqsTransportConfig
# Configure transport
config = SnsSqsTransportConfig(
sqs_queue_url="https://sqs.region.amazonaws.com/account/client-queue",
sns_topic_arn="arn:aws:sns:region:account:mcp-requests"
) # more configurable params available.
# Create AWS clients
sqs_client = boto3.client('sqs')
sns_client = boto3.client('sns')
async def main():
async with sns_sqs_client(config, sqs_client, sns_client) as (read_stream, write_stream):
# Your MCP client logic here
pass
```
You can check full examples at `/examples/website_server.py` and `/examples/website_client.py`.
<br/>
Read more at `/examples/README.md`
## Limitations
- **Message Size**: For SQS - message size limits are applicable (256KB standard, 2MB extended)
- **Response Handling**: Async nature means responses may not be immediate
- **Session Context**: Storage mechanism handled by server application, not transport
- **Ordering**: Standard SQS doesn't guarantee message ordering
## Testing
### Unit Tests
```bash
uv run pytest
```
## Contributing
We welcome contributions and discussions about async MCP architectures!
### Development Setup
```bash
git clone https://github.com/bharatgeleda/asyncmcp.git
cd asyncmcp
uv sync
```
---
## License
Apache License 2.0 - see [LICENSE](LICENSE) file for details.
## Links
- **MCP Specification**: [https://spec.modelcontextprotocol.io](https://spec.modelcontextprotocol.io)
Raw data
{
"_id": null,
"home_page": null,
"name": "asyncmcp",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "async, automation, aws, llm, mcp, queue, sns, sqs",
"author": null,
"author_email": "Bharat Geleda <bharatgeleda@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/14/95/fc559cc1e598f9b408ba87237cdb28d791a2aa29cb7b4df55b00fb129502/asyncmcp-0.1.2.tar.gz",
"platform": null,
"description": "# asyncmcp - Async transport layers for MCP \n\n\n[](LICENSE)\n[](https://www.python.org/downloads/)\n\n\n---\n\n## Overview\n\n\nA regular MCP Server but working over queues :\n\nhttps://github.com/user-attachments/assets/4b775ff8-02ae-4730-a822-3e1cedf9d744\n\n\nQuoting from the [official description](https://modelcontextprotocol.io/introduction) :<br/> \n> MCP is an open protocol that standardizes how applications provide context to LLMs.\n\nBut a lot of this context is not always readily available and takes time for the applications to process - think batch processing APIs, webhooks or queues. In these cases with the current transport layers, the MCP server would have to expose a light-weight polling wrapper in the MCP layer to allow waiting and polling for the tasks to be done. Although SSE does provide async functionalities but it comes with caveats. <br/>\n\nasyncmcp explores supporting more of the async transport layer implementations for MCP clients and servers, beyond the officially supported stdio and Streamable Http transports. \n\nThe whole idea of an **MCP server with async transport layer** is that it doesn't have to respond immediately to any requests. It can choose to direct them to internal queues for processing and the client doesn't have to stick around for the response.\n\n## Current capabilities\n\n### Transport layer : sns-sqs\n\n- Server : Transport layer that listens to a queue for MCP requests and writes the responses to a topic\n- Client : Transport layer that writes requests to a topic and listens to a queue for responses\n\n## Installation and Usage\n\n```bash\n# Using uv (recommended)\nuv add asyncmcp\n```\n\n```bash\n# Using pip \npip install asyncmcp\n```\n\n### Basic server setup\n\nNote : we don't support FastMCP yet. The examples in this repo uses the [basic way of creating MCP servers and client](https://modelcontextprotocol.io/docs/concepts/architecture#implementation-example) \n<br/>\nHere's a basic example of implementing an MCP server which supports sns-sqs as the transport layer:\n\n```python\nimport boto3\nfrom asyncmcp.sns_sqs.server import sns_sqs_server\nfrom asyncmcp import SnsSqsTransportConfig\n\n# Configure transport\nconfig = SnsSqsTransportConfig(\n sqs_queue_url=\"https://sqs.region.amazonaws.com/account/service-queue\",\n sns_topic_arn=\"arn:aws:sns:region:account:mcp-responses\"\n) # more configurable params available.\n\n# Create AWS clients\nsqs_client = boto3.client('sqs')\nsns_client = boto3.client('sns')\n\nasync def main():\n async with sns_sqs_server(config, sqs_client, sns_client) as (read_stream, write_stream):\n # Your MCP server logic here\n pass\n```\n\n### Basic client setup\n\nHere's a basic example of implementing an MCP client which supports sns-sqs as the transport layer:\n\n```python\nimport boto3\nfrom asyncmcp.sns_sqs.client import sns_sqs_client\nfrom asyncmcp import SnsSqsTransportConfig\n\n# Configure transport\nconfig = SnsSqsTransportConfig(\n sqs_queue_url=\"https://sqs.region.amazonaws.com/account/client-queue\",\n sns_topic_arn=\"arn:aws:sns:region:account:mcp-requests\"\n) # more configurable params available.\n\n# Create AWS clients\nsqs_client = boto3.client('sqs')\nsns_client = boto3.client('sns')\n\nasync def main():\n async with sns_sqs_client(config, sqs_client, sns_client) as (read_stream, write_stream):\n # Your MCP client logic here\n pass\n```\n\nYou can check full examples at `/examples/website_server.py` and `/examples/website_client.py`. \n<br/>\nRead more at `/examples/README.md`\n\n## Limitations\n\n- **Message Size**: For SQS - message size limits are applicable (256KB standard, 2MB extended)\n- **Response Handling**: Async nature means responses may not be immediate\n- **Session Context**: Storage mechanism handled by server application, not transport\n- **Ordering**: Standard SQS doesn't guarantee message ordering\n\n## Testing\n\n### Unit Tests\n\n```bash\nuv run pytest\n```\n\n\n\n## Contributing\n\nWe welcome contributions and discussions about async MCP architectures!\n\n### Development Setup\n\n```bash\ngit clone https://github.com/bharatgeleda/asyncmcp.git\ncd asyncmcp\nuv sync\n```\n\n---\n\n## License\n\nApache License 2.0 - see [LICENSE](LICENSE) file for details.\n\n## Links\n\n- **MCP Specification**: [https://spec.modelcontextprotocol.io](https://spec.modelcontextprotocol.io)\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "Async MCP Transport layer for queue and async based systems",
"version": "0.1.2",
"project_urls": {
"Bug Tracker": "https://github.com/bh-rat/asyncmcp/issues",
"Changelog": "https://github.com/bh-rat/asyncmcp/blob/main/CHANGELOG.md",
"Documentation": "https://github.com/bh-rat/asyncmcp#readme",
"Homepage": "https://github.com/bh-rat/asyncmcp",
"Repository": "https://github.com/bh-rat/asyncmcp"
},
"split_keywords": [
"async",
" automation",
" aws",
" llm",
" mcp",
" queue",
" sns",
" sqs"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "ddd955dcac313849fbd656c80d535727b4ae6700630762909bf8e1bc6bcca1d8",
"md5": "8e082e02b8f037bfe049265cc2012515",
"sha256": "65eacb2116f5ae2224c4159a0573f252339038737264d8a087aedeb6142d00dc"
},
"downloads": -1,
"filename": "asyncmcp-0.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "8e082e02b8f037bfe049265cc2012515",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 11811,
"upload_time": "2025-07-10T16:30:07",
"upload_time_iso_8601": "2025-07-10T16:30:07.635009Z",
"url": "https://files.pythonhosted.org/packages/dd/d9/55dcac313849fbd656c80d535727b4ae6700630762909bf8e1bc6bcca1d8/asyncmcp-0.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "1495fc559cc1e598f9b408ba87237cdb28d791a2aa29cb7b4df55b00fb129502",
"md5": "141891a144948eff325c74b8da1d7366",
"sha256": "8d674dc477a6046a5000a5fceff65c75bd03fdfa2447a39222c3cacf10528aba"
},
"downloads": -1,
"filename": "asyncmcp-0.1.2.tar.gz",
"has_sig": false,
"md5_digest": "141891a144948eff325c74b8da1d7366",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 21301,
"upload_time": "2025-07-10T16:30:08",
"upload_time_iso_8601": "2025-07-10T16:30:08.586345Z",
"url": "https://files.pythonhosted.org/packages/14/95/fc559cc1e598f9b408ba87237cdb28d791a2aa29cb7b4df55b00fb129502/asyncmcp-0.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-10 16:30:08",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "bh-rat",
"github_project": "asyncmcp",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "asyncmcp"
}
JSON
