# mcpbytes-lambda-core
[](https://www.python.org/downloads/)
[](https://spec.modelcontextprotocol.io/)
[](https://opensource.org/licenses/Apache-2.0)
Transport-agnostic [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) server core for AWS Lambda. Build production-ready MCP servers that work across multiple transports without vendor lock-in.
## π **Quick Start**
```python
from mcpbytes_lambda.core import MCPServer, ToolResult, TextContent
# Create server with default 1MB payload limit
mcp = MCPServer(name="math-server", version="1.0.0")
@mcp.tool(name="math.add", read_only=True, idempotent=True)
def add_numbers(a: float, b: float) -> ToolResult:
"""Add two numbers with high precision."""
result = a + b
return ToolResult(
content=[TextContent(text=f"{a} + {b} = {result}")],
isError=False
)
# Use with any transport (API Gateway, stdio, etc.)
def lambda_handler(event, context):
from mcpbytes_lambda.apigw import ApiGatewayAdapter
return mcp.handle(event, ApiGatewayAdapter())
```
## βοΈ **Payload Size Configuration**
Configure maximum request payload size based on your transport:
```python
# Default (1MB) - works across all transports
mcp = MCPServer(name="my-server")
# API Gateway optimized (10MB maximum)
mcp = MCPServer(name="api-server", max_payload_size=10_000_000)
# Stdio unlimited (for local development)
mcp = MCPServer(name="stdio-server", max_payload_size=100_000_000)
# Conservative (512KB for ALB)
mcp = MCPServer(name="alb-server", max_payload_size=512_000)
```
**Transport Limits:**
- **API Gateway**: 10MB maximum
- **Lambda Direct**: 6MB maximum
- **ALB**: 1MB maximum
- **Stdio**: No inherent limit
## β¨ **Features**
- **π Transport Agnostic** - Works with API Gateway, stdio, Lambda Function URLs, ALB
- **π Auto Schema Generation** - JSON Schema from Python type hints
- **π‘οΈ Built-in Validation** - Parameter validation with clear error messages
- **β‘ Sync + Async Support** - Handle both synchronous and asynchronous tools
- **π Structured Output** - MCP 2025-06-18 structured data support
- **π― Zero Cold Start** - Optimized for AWS Lambda performance
- **π Production Ready** - Comprehensive error handling and logging
## π **Requirements**
- Python 3.12+
- No runtime dependencies (pure Python)
## ποΈ **Architecture**
The core provides the foundation for MCP servers without transport coupling:
```
βββββββββββββββββββ ββββββββββββββββ βββββββββββββββββββ
β Transport βββββΆβ Core βββββΆβ Your Tools β
β (HTTP/stdio) β β Protocol β β & Prompts β
βββββββββββββββββββ ββββββββββββββββ βββββββββββββββββββ
```
## π οΈ **Usage Examples**
### **Simple Tool**
```python
@mcp.tool(name="greet", description="Greet a user")
def greet(name: str) -> ToolResult:
return ToolResult(
content=[TextContent(text=f"Hello, {name}!")],
isError=False
)
```
### **Structured Output Tool** (MCP 2025-06-18)
```python
@mcp.tool(name="calculate", structured_output=True)
def calculate(operation: str, values: List[float]) -> Dict[str, Any]:
"""Returns structured data with auto-generated schema."""
result = sum(values) if operation == "sum" else max(values)
return {
"operation": operation,
"input_values": values,
"result": result,
"timestamp": "2025-06-18T12:00:00Z"
}
```
### **Async Tool**
```python
@mcp.tool(name="fetch_data", read_only=True)
async def fetch_data(url: str) -> ToolResult:
# Async operations supported natively
async with httpx.AsyncClient() as client:
response = await client.get(url)
return ToolResult(
content=[TextContent(text=response.text[:500])],
isError=False
)
```
### **Prompt Templates**
```python
@mcp.prompt(name="code_review")
def code_review_prompt(language: str, code: str) -> str:
"""Generate code review prompt."""
return f"Review this {language} code for best practices:\n\n```{language}\n{code}\n```"
```
## π **Transport Adapters**
The core works with multiple transport adapters:
- **`mcpbytes-lambda-apigw`** - API Gateway + Lambda
- **`mcpbytes-lambda-stdio`** - stdin/stdout (local MCP servers)
- **`mcpbytes-lambda-invoke`** - Direct Lambda invocation
- **`mcpbytes-lambda-alb`** - Application Load Balancer + Lambda
## π **API Reference**
### **MCPServer**
```python
MCPServer(
name: str, # Server name
version: str = "1.0.0", # Server version
description: Optional[str] = None, # Server description
log_level: str = "INFO", # Logging level
max_payload_size: int = 1_000_000 # Maximum request payload size in bytes
)
```
### **@tool Decorator**
```python
@mcp.tool(
name: Optional[str] = None, # Tool name (defaults to function name)
description: Optional[str] = None, # Description (from docstring)
title: Optional[str] = None, # Human-readable title
read_only: bool = False, # Tool only reads data
destructive: bool = False, # Tool can destroy data
idempotent: bool = False, # Safe to call repeatedly
open_world: bool = True, # Interacts with external systems
structured_output: bool = False # Enable structured output (MCP 2025-06-18)
)
```
### **ToolResult**
```python
ToolResult(
content: List[ContentBlock], # Text/resource content blocks
isError: bool = False, # Whether this is an error
structuredContent: Optional[Dict] = None # Structured data (MCP 2025-06-18)
)
```
## π§ͺ **Testing**
```python
# Test your tools directly
def test_add_tool():
result = add_numbers(2, 3)
assert not result.isError
assert "5" in result.content[0].text
# Test with mock transport
def test_server_integration():
event = {"body": '{"jsonrpc":"2.0","method":"tools/list","id":"1"}'}
from mcpbytes_lambda.apigw import ApiGatewayAdapter
response = mcp.handle(event, ApiGatewayAdapter())
assert response["statusCode"] == 200
```
## π€ **Contributing**
1. Fork the repository
2. Create a feature branch: `git checkout -b feature-name`
3. Make your changes and add tests
4. Run tests: `python -m pytest`
5. Submit a pull request
## π **License**
Apache 2.0 License - see [LICENSE](../../LICENSE) for details.
## π **Related Packages**
- [`mcpbytes-lambda-apigw`](../apigw/) - API Gateway transport adapter
- [`mcpbytes-lambda-stdio`](../stdio/) - Stdio transport adapter
- [`mcpbytes-lambda-invoke`](../invoke/) - Direct Lambda invocation adapter
## π **Documentation**
- [MCP Specification](https://spec.modelcontextprotocol.io/)
- [AWS Lambda Documentation](https://docs.aws.amazon.com/lambda/)
- [Project Examples](../../examples/)
---
Built with β€οΈ for the MCP ecosystem
Raw data
{
"_id": null,
"home_page": null,
"name": "mcpbytes-lambda-core",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.12",
"maintainer_email": null,
"keywords": "ai, aws-lambda, core, mcp, model-context-protocol, serverless",
"author": null,
"author_email": "\"MCPBytes.com\" <hello@mcpbytes.com>",
"download_url": "https://files.pythonhosted.org/packages/d1/69/f327d223b11e672f885c06de0caefffaa043e9c3a9f571729820b4f62873/mcpbytes_lambda_core-0.2.1.tar.gz",
"platform": null,
"description": "# mcpbytes-lambda-core\n\n[](https://www.python.org/downloads/)\n[](https://spec.modelcontextprotocol.io/)\n[](https://opensource.org/licenses/Apache-2.0)\n\nTransport-agnostic [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) server core for AWS Lambda. Build production-ready MCP servers that work across multiple transports without vendor lock-in.\n\n## \ud83d\ude80 **Quick Start**\n\n```python\nfrom mcpbytes_lambda.core import MCPServer, ToolResult, TextContent\n\n# Create server with default 1MB payload limit\nmcp = MCPServer(name=\"math-server\", version=\"1.0.0\")\n\n@mcp.tool(name=\"math.add\", read_only=True, idempotent=True)\ndef add_numbers(a: float, b: float) -> ToolResult:\n \"\"\"Add two numbers with high precision.\"\"\"\n result = a + b\n return ToolResult(\n content=[TextContent(text=f\"{a} + {b} = {result}\")],\n isError=False\n )\n\n# Use with any transport (API Gateway, stdio, etc.)\ndef lambda_handler(event, context):\n from mcpbytes_lambda.apigw import ApiGatewayAdapter\n return mcp.handle(event, ApiGatewayAdapter())\n```\n\n## \u2699\ufe0f **Payload Size Configuration**\n\nConfigure maximum request payload size based on your transport:\n\n```python\n# Default (1MB) - works across all transports\nmcp = MCPServer(name=\"my-server\")\n\n# API Gateway optimized (10MB maximum)\nmcp = MCPServer(name=\"api-server\", max_payload_size=10_000_000)\n\n# Stdio unlimited (for local development)\nmcp = MCPServer(name=\"stdio-server\", max_payload_size=100_000_000)\n\n# Conservative (512KB for ALB)\nmcp = MCPServer(name=\"alb-server\", max_payload_size=512_000)\n```\n\n**Transport Limits:**\n- **API Gateway**: 10MB maximum\n- **Lambda Direct**: 6MB maximum \n- **ALB**: 1MB maximum\n- **Stdio**: No inherent limit\n\n## \u2728 **Features**\n\n- **\ud83d\udd0c Transport Agnostic** - Works with API Gateway, stdio, Lambda Function URLs, ALB\n- **\ud83d\udcdd Auto Schema Generation** - JSON Schema from Python type hints\n- **\ud83d\udee1\ufe0f Built-in Validation** - Parameter validation with clear error messages\n- **\u26a1 Sync + Async Support** - Handle both synchronous and asynchronous tools\n- **\ud83d\udcca Structured Output** - MCP 2025-06-18 structured data support\n- **\ud83c\udfaf Zero Cold Start** - Optimized for AWS Lambda performance\n- **\ud83d\udd12 Production Ready** - Comprehensive error handling and logging\n\n## \ud83d\udccb **Requirements**\n\n- Python 3.12+\n- No runtime dependencies (pure Python)\n\n## \ud83c\udfd7\ufe0f **Architecture**\n\nThe core provides the foundation for MCP servers without transport coupling:\n\n```\n\u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510 \u250c\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2510\n\u2502 Transport \u2502\u2500\u2500\u2500\u25b6\u2502 Core \u2502\u2500\u2500\u2500\u25b6\u2502 Your Tools \u2502\n\u2502 (HTTP/stdio) \u2502 \u2502 Protocol \u2502 \u2502 & Prompts \u2502\n\u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518 \u2514\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2518\n```\n\n## \ud83d\udee0\ufe0f **Usage Examples**\n\n### **Simple Tool**\n\n```python\n@mcp.tool(name=\"greet\", description=\"Greet a user\")\ndef greet(name: str) -> ToolResult:\n return ToolResult(\n content=[TextContent(text=f\"Hello, {name}!\")],\n isError=False\n )\n```\n\n### **Structured Output Tool** (MCP 2025-06-18)\n\n```python\n@mcp.tool(name=\"calculate\", structured_output=True)\ndef calculate(operation: str, values: List[float]) -> Dict[str, Any]:\n \"\"\"Returns structured data with auto-generated schema.\"\"\"\n result = sum(values) if operation == \"sum\" else max(values)\n return {\n \"operation\": operation,\n \"input_values\": values,\n \"result\": result,\n \"timestamp\": \"2025-06-18T12:00:00Z\"\n }\n```\n\n### **Async Tool**\n\n```python\n@mcp.tool(name=\"fetch_data\", read_only=True)\nasync def fetch_data(url: str) -> ToolResult:\n # Async operations supported natively\n async with httpx.AsyncClient() as client:\n response = await client.get(url)\n return ToolResult(\n content=[TextContent(text=response.text[:500])],\n isError=False\n )\n```\n\n### **Prompt Templates**\n\n```python\n@mcp.prompt(name=\"code_review\")\ndef code_review_prompt(language: str, code: str) -> str:\n \"\"\"Generate code review prompt.\"\"\"\n return f\"Review this {language} code for best practices:\\n\\n```{language}\\n{code}\\n```\"\n```\n\n## \ud83d\udd0c **Transport Adapters**\n\nThe core works with multiple transport adapters:\n\n- **`mcpbytes-lambda-apigw`** - API Gateway + Lambda\n- **`mcpbytes-lambda-stdio`** - stdin/stdout (local MCP servers) \n- **`mcpbytes-lambda-invoke`** - Direct Lambda invocation\n- **`mcpbytes-lambda-alb`** - Application Load Balancer + Lambda\n\n## \ud83d\udcd6 **API Reference**\n\n### **MCPServer**\n\n```python\nMCPServer(\n name: str, # Server name\n version: str = \"1.0.0\", # Server version \n description: Optional[str] = None, # Server description\n log_level: str = \"INFO\", # Logging level\n max_payload_size: int = 1_000_000 # Maximum request payload size in bytes\n)\n```\n\n### **@tool Decorator**\n\n```python\n@mcp.tool(\n name: Optional[str] = None, # Tool name (defaults to function name)\n description: Optional[str] = None, # Description (from docstring)\n title: Optional[str] = None, # Human-readable title\n read_only: bool = False, # Tool only reads data\n destructive: bool = False, # Tool can destroy data\n idempotent: bool = False, # Safe to call repeatedly\n open_world: bool = True, # Interacts with external systems\n structured_output: bool = False # Enable structured output (MCP 2025-06-18)\n)\n```\n\n### **ToolResult**\n\n```python\nToolResult(\n content: List[ContentBlock], # Text/resource content blocks\n isError: bool = False, # Whether this is an error\n structuredContent: Optional[Dict] = None # Structured data (MCP 2025-06-18)\n)\n```\n\n## \ud83e\uddea **Testing**\n\n```python\n# Test your tools directly\ndef test_add_tool():\n result = add_numbers(2, 3)\n assert not result.isError\n assert \"5\" in result.content[0].text\n\n# Test with mock transport\ndef test_server_integration():\n event = {\"body\": '{\"jsonrpc\":\"2.0\",\"method\":\"tools/list\",\"id\":\"1\"}'}\n from mcpbytes_lambda.apigw import ApiGatewayAdapter\n \n response = mcp.handle(event, ApiGatewayAdapter())\n assert response[\"statusCode\"] == 200\n```\n\n## \ud83e\udd1d **Contributing**\n\n1. Fork the repository\n2. Create a feature branch: `git checkout -b feature-name`\n3. Make your changes and add tests\n4. Run tests: `python -m pytest`\n5. Submit a pull request\n\n## \ud83d\udcc4 **License**\n\nApache 2.0 License - see [LICENSE](../../LICENSE) for details.\n\n## \ud83d\udd17 **Related Packages**\n\n- [`mcpbytes-lambda-apigw`](../apigw/) - API Gateway transport adapter\n- [`mcpbytes-lambda-stdio`](../stdio/) - Stdio transport adapter \n- [`mcpbytes-lambda-invoke`](../invoke/) - Direct Lambda invocation adapter\n\n## \ud83d\udcda **Documentation**\n\n- [MCP Specification](https://spec.modelcontextprotocol.io/)\n- [AWS Lambda Documentation](https://docs.aws.amazon.com/lambda/)\n- [Project Examples](../../examples/)\n\n---\n\nBuilt with \u2764\ufe0f for the MCP ecosystem\n",
"bugtrack_url": null,
"license": null,
"summary": "Transport-agnostic Model Context Protocol (MCP) server core for AWS Lambda",
"version": "0.2.1",
"project_urls": {
"Documentation": "https://github.com/MCPBytes/mcpbytes-lambda#readme",
"Homepage": "https://mcpbytes.com/",
"Issues": "https://github.com/MCPBytes/mcpbytes-lambda/issues",
"Repository": "https://github.com/MCPBytes/mcpbytes-lambda"
},
"split_keywords": [
"ai",
" aws-lambda",
" core",
" mcp",
" model-context-protocol",
" serverless"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "552f8f7d5a08e90f0c6fc0ad844297a27efd4a0db7e5fc7307f774a7fd631092",
"md5": "d380ca194936b64cea7e5436d3c329b9",
"sha256": "6f987e8ddd39dec028aac2087ed162a4c39f8bb1b862a8c0b62bf79951d567f3"
},
"downloads": -1,
"filename": "mcpbytes_lambda_core-0.2.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "d380ca194936b64cea7e5436d3c329b9",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12",
"size": 22121,
"upload_time": "2025-09-01T20:05:59",
"upload_time_iso_8601": "2025-09-01T20:05:59.475917Z",
"url": "https://files.pythonhosted.org/packages/55/2f/8f7d5a08e90f0c6fc0ad844297a27efd4a0db7e5fc7307f774a7fd631092/mcpbytes_lambda_core-0.2.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "d169f327d223b11e672f885c06de0caefffaa043e9c3a9f571729820b4f62873",
"md5": "29a46306b3b2eab27966d06dd74809d2",
"sha256": "eb953f574b1f9c8a2249e4250d911c2272174753e6a5b6bd04208a5606caa7cd"
},
"downloads": -1,
"filename": "mcpbytes_lambda_core-0.2.1.tar.gz",
"has_sig": false,
"md5_digest": "29a46306b3b2eab27966d06dd74809d2",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12",
"size": 49044,
"upload_time": "2025-09-01T20:06:00",
"upload_time_iso_8601": "2025-09-01T20:06:00.791571Z",
"url": "https://files.pythonhosted.org/packages/d1/69/f327d223b11e672f885c06de0caefffaa043e9c3a9f571729820b4f62873/mcpbytes_lambda_core-0.2.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-09-01 20:06:00",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "MCPBytes",
"github_project": "mcpbytes-lambda#readme",
"github_not_found": true,
"lcname": "mcpbytes-lambda-core"
}