AWS Model Context Protocol (MCP) Server
The AWS API Model Context Protocol (MCP) Server provides a Python implementation of the gRPC-based Model Context Protocol. It allows an external client to provide API models and context to the server, assisting in API construction. It is currently at version 1.3.29 and receives regular updates, primarily for patch and minor feature releases.
Warnings
- gotcha The server runs as an asynchronous gRPC service. Ensure your custom `ModelContextServiceServicer` implementations are also `async def` functions and correctly handle asynchronous operations. Blocking calls within these methods can degrade performance.
- breaking Major changes to the underlying Model Context Protocol definition (e.g., in `mcp.proto`) could introduce breaking changes to the gRPC service interface. While `awslabs` strives for backward compatibility, new major versions of the library might reflect such protocol updates.
- gotcha Correctly loading and managing API models and context is crucial for the server's function. The example provides a basic `AddContext`, but production implementations will require robust storage and retrieval mechanisms for the models and their associated context.
- gotcha The library relies on specific versions of `grpcio` and `protobuf`. Incompatibility issues can arise if other projects in your environment use different or conflicting versions of these dependencies.
Install
-
pip install awslabs-aws-api-mcp-server
Imports
- McpServer
from mcp.server import McpServer
- ModelContextServiceServicer
from mcp.service import ModelContextServiceServicer
- AddContextRequest
from mcp.server.models.mcp_pb2 import AddContextRequest
Quickstart
import asyncio
from concurrent import futures
import grpc
from mcp.server import McpServer
from mcp.service import ModelContextServiceServicer
from mcp.server.models.mcp_pb2_grpc import add_ModelContextServiceServicer_to_server
from mcp.server.models.mcp_pb2 import AddContextRequest, AddContextResponse
class MyMcpService(ModelContextServiceServicer):
"""Example implementation of the ModelContextServiceServicer."""
async def AddContext(self, request: AddContextRequest, context: grpc.aio.ServicerContext) -> AddContextResponse:
print(f"Received AddContext request for model: {request.model_name}")
# Implement actual context logic here
return AddContextResponse(success=True, message="Context added successfully")
async def GetContext(self, request, context):
# Implement GetContext logic
return super().GetContext(request, context)
async def serve():
server = grpc.aio.server(futures.ThreadPoolExecutor(max_workers=10))
add_ModelContextServiceServicer_to_server(MyMcpService(), server)
server.add_insecure_port('[::]:50051')
print('Starting MCP server on port 50051')
await server.start()
await server.wait_for_termination()
if __name__ == '__main__':
asyncio.run(serve())