MCP Lambda服务器模板

一个可重复使用的AWS SAM模板和参考实现,用于在AWS Lambda和API Gateway上部署具有基于令牌授权的模型上下文协议(MCP)服务器。

  • by VPPATEL-SPS
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 vppatel-sps/mcp-lambda-server

使用说明

通用MCP Lambda服务器模板

可重复使用的AWS SAM模板和参考实现,用于在AWS Lambda和API网关上部署具有基于令牌授权的模型上下文协议(MCP)服务器。使用此通用蓝图构建和扩展您自己的MCP工具,根据需要集成自定义业务逻辑、数据源或外部API。

✨ 包含什么

  • AWS SAM/CloudFormation基础设施(API网关、Lambda、可选DynamoDB表、会话存储)
  • 可插拔令牌授权器(简单承载令牌)
  • 会话持久性表(基于TTL)
  • 示例工具(echo、add、utc_timestamp、可选DynamoDB示例)
  • 最小化依赖足迹

🗂 项目结构

mcp-lambda-server/
├── template.yml                # SAM infrastructure definition
├── README.md                   # This guide
├── authorizer/
│   ├── app.py                  # Custom Lambda token authorizer
├── server/
│   ├── app.py                  # MCP server + example tools
│   └── requirements.txt        # Python dependencies
└── examples/
    ├── simple_client.py        # Minimal MCP call example
    └── strands_client.py       # Interactive Strands Agent chat example

🧰 提供的示例工具

工具 目的
echo(text) 返回相同的文本(连接健全性检查)
add(a,b) 将两个数字相加并返回结构化JSON结果
utc_timestamp() 返回当前utc时间戳和ISO8601字符串
list_items(limit) (可选)列出演示DynamoDB表中的项目(如果已配置)
health() 基本服务器/配置状态

您可以自由删除或扩展这些。

🏗 建筑

┌─────────────┐   HTTPS (Bearer token)   ┌──────────────┐
│ MCP Client  │ ───────────────────────▶ │ API Gateway  │
└─────────────┘                          │  (Authorizer)│
                                         └──────┬───────┘
                                                │Invoke
                                         ┌───────▼────────┐
                                         │  Lambda (MCP)  │
                                         └───────┬────────┘
                                                 │ (optional)
                                         ┌───────▼────────┐
                                         │ DynamoDB Table │
                                         └────────────────┘

🔐 安全模型

  • 简单的承载令牌(传递为 Authorization: Bearer <token>)由Lambda授权人验证。
  • 令牌存储为NoEcho CloudFormation参数(不以明文形式记录)。
  • 会话表使用TTL进行自动清理。

⚙ 参数(template.yml)

名称 描述 默认值 必填
McpAuthToken 承载令牌客户端必须在此处发送 您的安全令牌 是(产品更改)
CreateDemoTable 为list_items工具创建示例DynamoDB表 false 否
DemoTableName 现有表或新表的名称(如果CreateDemoTable=true) McpDemoItems 有条件

🧪 本地开发(选项A:SAM CLI)

# From repo root (where template.yml resides)
cd mcp-lambda-server
sam build
sam local invoke McpServerFunction --event events/sample.json

您可以创建一个最小调用事件(POST主体)——请参阅 awslabs.mcp_lambda_handler 请求模式的文档。通常,您将通过HTTP进行部署和调用。

🚀 部署

cd mcp-lambda-server
sam build
sam deploy \
  --stack-name generic-mcp-server \
  --capabilities CAPABILITY_IAM \
  --parameter-overrides \
    McpAuthToken="REPLACE_WITH_STRONG_TOKEN" \
    CreateDemoTable="true" \
    DemoTableName="McpDemoItems"

部署后注意输出 McpApiUrl例如:

https://xxxxxxxxxx.execute-api.us-east-1.amazonaws.com/Prod/mcp

🔌 从GitHub Copilot连接(mcp.json)

示例条目:

{
  "servers": {
    "generic-mcp": {
      "url": "https://xxxxxxxxxx.execute-api.us-east-1.amazonaws.com/Prod/mcp",
      "requestInit": {
        "headers": { "Authorization": "Bearer REPLACE_WITH_STRONG_TOKEN" }
      }
    }
  }
}

➕ 添加新工具

在……里面 server/app.py:

@mcp_server.tool()
def my_new_tool(arg1: str) -> str:
    return json.dumps({"tool": "my_new_tool", "arg1": arg1})

返回纯字符串或JSON序列化字符串。保持回答简洁明了。

🧱 DynamoDB演示表架构(可选)

如果 CreateDemoTable=true,一个带有分区键的PAY_PER_REQUEST表 pk 创建。您可以手动插入示例项:

aws dynamodb put-item \
  --table-name McpDemoItems \
  --item '{"pk":{"S":"item#1"},"message":{"S":"Hello"}}'

然后打电话 list_items(limit=25) 通过MCP客户端。

🧪 快速MCP客户端冒烟测试(Python)

from mcp.client.streamable_http import streamablehttp_client
from strands.tools.mcp import MCPClient

API_URL = "https://xxxxxxxxxx.execute-api.us-east-1.amazonaws.com/Prod/mcp"
TOKEN = "REPLACE_WITH_STRONG_TOKEN"

client = MCPClient(lambda: streamablehttp_client(
    url=API_URL,
    headers={"Authorization": f"Bearer {TOKEN}"}
))

with client:
    tools = client.list_tools_sync()
    print("Tools:", [t.name for t in tools])
    res = client.call_tool_sync("utc_timestamp", {})
    print(res.content[0].text)

👩‍💻 最小直接使用示例

(参见 examples/simple_client.py)

from mcp.client.streamable_http import streamablehttp_client
from strands.tools.mcp import MCPClient
import json

API_URL = "https://your-api.execute-api.us-east-1.amazonaws.com/Prod/mcp"
TOKEN = "YOUR_TOKEN"

client = MCPClient(lambda: streamablehttp_client(
    url=API_URL,
    headers={"Authorization": f"Bearer {TOKEN}"}
))

with client:
    for t in client.list_tools_sync():
        print("Tool:", t.name, t.description)
    # Call echo tool
    resp = client.call_tool_sync("echo", {"text": "hello"})
    print("echo →", resp.content[0].text)

🗣 交互式Strands代理聊天示例

(参见 examples/strands_client.py)使用基岩模型+发现的MCP工具。

关键步骤:

  1. 使用承载身份验证构建MCP客户端
  2. 列出工具
  3. 构建代理(模型=…,工具=工具)
  4. 读取用户输入循环并调用 agent(user_text)

环境要求:

  • strands (及其基岩整合)和依赖关系
  • 具有区域访问权限的正确基岩模型ID

📁 示例脚本摘要

文件 目的
examples/simple_client.py 最小列表+调用工具演示
examples/strands_client.py 与strands Agent的交互式聊天循环

🩺 健康与故障排除

症状 检查
403未经授权 确保标头使用正确的承载令牌并与部署的参数匹配
500错误(工具) CloudWatch日志 McpServerFunction 用于堆栈跟踪
超时 增加Lambda超时 Globals.Function.Timeout
空工具列表 确认正确的URL路径 /mcp POST方法

快速获取日志:

sam logs -n McpServerFunction --stack-name generic-mcp-server --tail

🧹 维护

  • 更新代码→ sam build && sam deploy
  • 旋转令牌→ 部署新参数,更新所有客户端
  • 添加工具→ 修改 server/app.py 并重新部署

📄 许可证

在您的组织中自由使用;在生产前验证安全态势。


请随时根据您的特定域逻辑(Jira、内部API等)进一步定制此模板。