使用 AWS Lambda 运行模式上下文协议服务器
此项目允许您运行 示范背景协议 aWS Lambda 函数中基于stdio的服务器.
目前,大多数MCP服务器和客户端的应用都是完全本地化的单机. 桌面应用程序, 如 IDE 或 Claude Desktop 在本地启动 MCP 服务器作为子进程 并通过一个长期运行的 stdio流与每个服务器通信.
flowchart LR
subgraph "Your Laptop"
Host["Desktop Application<br>with MCP Clients"]
S1["MCP Server A<br>(child process)"]
S2["MCP Server B<br>(child process)"]
Host <-->|"MCP Protocol<br>(over stdio stream)"| S1
Host <-->|"MCP Protocol<br>(over stdio stream)"| S2
end
这个库帮助您将已有的stdio MCP服务器包入Lambda函数. 您可以使用 MCP 协议从您的应用程序中引用这些基于函数的 MCP 服务器 在短寿命的HTTPS连接. 然后您的应用程序可以是一个基于桌面的应用程序,一个在云中运行的分布式系统, 或其他建筑.
flowchart LR
subgraph "Distributed System"
App["Your Application<br>with MCP Clients"]
S3["MCP Server A<br>(Lambda function)"]
S4["MCP Server B<br>(Lambda function)"]
App <-->|"MCP Protocol<br>(over HTTPS connection)"| S3
App <-->|"MCP Protocol<br>(over HTTPS connection)"| S4
end
使用这个库,Lambda函数将管理您stdio MCP服务器的生命周期. 每个Lambda函数引用将:
- 启动 stdio MCP 服务器作为子进程
- 初始化 MCP 服务器
- 将收到的请求转发到本地服务器
- 返回服务器对函数调用器的响应
- 关闭 MCP 服务器子进程
这个库支持连接到基于Lambda的MCP服务器有四种方式:
- 这个 MCP 可流式HTTP运输,使用亚马逊API Gateway. 通常使用 OAuth 认证 .
- MCP可流式HTTP运输,使用亚马逊Bedrock Agent Core Gateway. 使用 OAuth 认证 .
- 支持SigV4的自定义流式HTTP运输,使用Lambda函数 URL. 以AWS IAM认证.
- 习惯上Lambda引用运输,直接使用Lambda Invoke API. 以AWS IAM认证.
确定您的服务器参数
许多基于stdio的MCP服务器的文档都鼓励使用下载和运行服务器按需运行的工具.
譬如说, uvx my-mcp-server 或者说 npx my-mcp-server。 。 。 .
这些工具往往没有在Lambda环境中预先包装,而且可能效率低下
每一次Lambda引用时 重新下载服务器.
相反,此寄存器中的例子显示如何将 MCP 服务器与
lambda 函数代码,然后以 python 或者说 node (或 npx --offline直接进行.
您需要根据您的 MCP 服务器的软件包来确定正确的参数 . 由于MCP服务器包装各有不同,这常常可能是本地的试验和出错过程.
Python 服务器实例
基本实例:
from mcp.client.stdio import StdioServerParameters
server_params = StdioServerParameters(
command=sys.executable,
args=[
"-m",
"my_mcp_server_python_module",
"--my-server-command-line-parameter",
"some_value",
],
)
在本地,您将使用下列方式运行此模块:
python -m my_mcp_server_python_module --my-server-command-line-parameter some_value
其他例子:
python -m mcpdoc.cli # Note the sub-module
python -c "from mcp_openapi_proxy import main; main()"
python -c "import asyncio; from postgres_mcp.server import main; asyncio.run(main())"
如果您使用 Lambda 层, 您还需要为 Python 子进程设置 PYTHONPATH :
lambda_paths = ["/opt/python"] + sys.path
env_config = {"PYTHONPATH": ":".join(lambda_paths)}
server_params = StdioServerParameters(
command=sys.executable,
args=[
"-c",
"from mcp_openapi_proxy import main; main()",
],
env=env_config,
)
Typescript 服务器实例
基本实例:
const serverParams = {
command: "npx",
args: [
"--offline",
"my-mcp-server-typescript-module",
"--my-server-command-line-parameter",
"some_value",
],
};
在本地,您将使用下列方式运行此模块:
npx --offline my-mcp-server-typescript-module --my-server-command-line-parameter some_value
其他例子:
node /var/task/node_modules/@ivotoby/openapi-mcp-server/bin/mcp-server.js
向 MCP 服务器传递证书和其他机密
这个库不提供管理包裹所需的任何秘密的外出机制 MCP服务器. 例如 GitHub MCP 服务器 编辑 勇敢地搜索 MCP 服务器 要求 API 密钥向第三方 API 提出请求. 您可以将这些 API 密钥配置为 加密环境变量 在 Lambda 函数的配置中,或从 Lambda 函数代码中的秘密管理器中取回(下面的例子). 然而,请注意,任何能够援引Lambda函数的人 将使用您的 API 密钥使用此功能来调用第三方 API . 我们建议使用 最低特权的IAM政策。 。 。 . 如果您使用像OAuth这样的基于身份的认证机制,您也可以存储和检索每个用户的API密钥,但此寄存器中没有执行实例.
Python 服务器实例从保密管理器获取 API 密钥
import sys
import boto3
from mcp.client.stdio import StdioServerParameters
# Retrieve API key from Secrets Manager
secrets_client = boto3.client("secretsmanager")
api_key = secrets_client.get_secret_value(SecretId="my-api-key-secret")["SecretString"]
server_params = StdioServerParameters(
command=sys.executable,
args=["-m", "my_mcp_server"],
env={
"API_KEY": api_key,
},
)
Typescript 服务器实例从保密管理器获取 API 密钥
import { SecretsManagerClient, GetSecretValueCommand } from "@aws-sdk/client-secrets-manager";
const secretsClient = new SecretsManagerClient({});
const secret = await secretsClient.send(
new GetSecretValueCommand({ SecretId: "my-api-key-secret" })
);
const apiKey = secret.SecretString;
const serverParams = {
command: "npx",
args: ["--offline", "my-mcp-server"],
env: {
API_KEY: apiKey,
},
};
如果您的 MCP 服务器需要调用 AWS API( 例如 AWS 的 MCP 服务器, (中文(简体) ), 您可以通过环境变量将 Lambda 函数的 AWS 证书传递给包裹的 MCP 服务器. 包裹的MCP服务器的孩子过程不会自动地继承Lambda执行角色的证书. 请注意,任何能够援引Lambda函数的人 然后可以使用函数的 AWS 证书,通过引用函数来调用 AWS APIS. 我们建议使用 最低特权的IAM政策。 。 。 .
通过环境变量使用 AWS 证书的 Python 服务器示例
import os
import sys
import boto3
from mcp.client.stdio import StdioServerParameters
# Get AWS credentials from Lambda execution role to pass to subprocess
session = boto3.Session()
credentials = session.get_credentials()
if credentials is None:
raise RuntimeError("Unable to retrieve AWS credentials from the execution environment")
resolved = credentials.get_frozen_credentials()
server_params = StdioServerParameters(
command=sys.executable,
args=["-m", "my_mcp_server"],
env={
"AWS_REGION": os.environ.get("AWS_REGION", "us-west-2"),
"AWS_DEFAULT_REGION": os.environ.get("AWS_REGION", "us-west-2"),
"AWS_ACCESS_KEY_ID": resolved.access_key,
"AWS_SECRET_ACCESS_KEY": resolved.secret_key,
"AWS_SESSION_TOKEN": resolved.token or "",
},
)
通过证书文件使用 AWS 证书的 Python 服务器示例
一些MCP服务器需要AWS配置文件,不支持通过环境变量传递的资质. 在这种情况下,您可以将证书写入一个文件,并将MCP服务器指向它.
import os
import sys
import boto3
from mcp.client.stdio import StdioServerParameters
# Get AWS credentials from Lambda execution role to pass to subprocess
session = boto3.Session()
credentials = session.get_credentials()
if credentials is None:
raise RuntimeError("Unable to retrieve AWS credentials from the execution environment")
resolved = credentials.get_frozen_credentials()
# Write credentials to disk as default profile
aws_dir = "/tmp/.aws"
os.makedirs(aws_dir, exist_ok=True)
with open(f"{aws_dir}/credentials", "w") as f:
f.write("[default]\n")
f.write(f"aws_access_key_id = {resolved.access_key}\n")
f.write(f"aws_secret_access_key = {resolved.secret_key}\n")
if resolved.token:
f.write(f"aws_session_token = {resolved.token}\n")
server_params = StdioServerParameters(
command=sys.executable,
args=["-m", "my_mcp_server"],
env={
"AWS_REGION": os.environ.get("AWS_REGION", "us-west-2"),
"AWS_DEFAULT_REGION": os.environ.get("AWS_REGION", "us-west-2"),
"AWS_SHARED_CREDENTIALS_FILE": f"{aws_dir}/credentials",
},
)
见一个可部署的完整实例 这里。 。 。 .
使用 API 网关
flowchart LR
App["MCP Client"]
T1["MCP Server<br>(Lambda function)"]
T2["API Gateway"]
T3["OAuth Server<br>(Cognito or similar)"]
App -->|"MCP Streamable<br>HTTP Transport"| T2
T2 -->|"Invoke"| T1
T2 -->|"Authorize"| T3
这个解决方案与大多数支持可流式HTTP运输的MCP客户端相兼容. 使用此架构的 MCP 服务器通常可用现成 Cursor,Cline,克洛德桌面等MCP相容的应用程序.
您可以为此解决方案选择您想要的 OAuth 服务器提供者 。 这方面的实例 寄存器使用 Amazon Cognitto, 或者您可以使用第三方提供者, 如 Okta 或 Auth0 与 API 网关自定义授权 .
Python 服务器实例
import sys
from mcp.client.stdio import StdioServerParameters
from mcp_lambda import APIGatewayProxyEventHandler, StdioServerAdapterRequestHandler
server_params = StdioServerParameters(
command=sys.executable,
args=[
"-m",
"my_mcp_server_python_module",
"--my-server-command-line-parameter",
"some_value",
],
)
request_handler = StdioServerAdapterRequestHandler(server_params)
event_handler = APIGatewayProxyEventHandler(request_handler)
def handler(event, context):
return event_handler.handle(event, context)
见一个可部署的完整实例 这里。 。 。 .
Typescript 服务器实例
import {
Handler,
Context,
APIGatewayProxyWithCognitoAuthorizerEvent,
APIGatewayProxyResult,
} from "aws-lambda";
import {
APIGatewayProxyEventHandler,
StdioServerAdapterRequestHandler,
} from "@aws/run-mcp-servers-with-aws-lambda";
const serverParams = {
command: "npx",
args: [
"--offline",
"my-mcp-server-typescript-module",
"--my-server-command-line-parameter",
"some_value",
],
};
const requestHandler = new APIGatewayProxyEventHandler(
new StdioServerAdapterRequestHandler(serverParams)
);
export const handler: Handler = async (
event: APIGatewayProxyWithCognitoAuthorizerEvent,
context: Context
): Promise<APIGatewayProxyResult> => {
return requestHandler.handle(event, context);
};
见一个可部署的完整实例 这里。 。 。 .
Python 客户端示例
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
# Create OAuth client provider here
async with streamablehttp_client(
url="https://abc123.execute-api.us-west-2.amazonaws.com/prod/mcp",
auth=oauth_client_provider,
) as (
read_stream,
write_stream,
_,
):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
tool_result = await session.call_tool("echo", {"message": "hello"})
将一个完整的例子作为聊天样本的一部分 这里。 。 。 .
打印客户端示例
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
const client = new Client(
{
name: "my-client",
version: "0.0.1",
},
{
capabilities: {
sampling: {},
},
}
);
// Create OAuth client provider here
const transport = new StreamableHTTPClientTransport(
"https://abc123.execute-api.us-west-2.amazonaws.com/prod/mcp",
{
authProvider: oauthProvider,
}
);
await client.connect(transport);
将一个完整的例子作为聊天样本的一部分 这里。 。 。 .
使用 Bedrock 代理 Core 网关
flowchart LR
App["MCP Client"]
T1["MCP Server<br>(Lambda function)"]
T2["Bedrock AgentCore Gateway"]
T3["OAuth Server<br>(Cognito or similar)"]
App -->|"MCP Streamable<br>HTTP Transport"| T2
T2 -->|"Invoke"| T1
T2 -->|"Authorize"| T3
这个解决方案与大多数支持可流式HTTP运输的MCP客户端相兼容. 使用此架构的 MCP 服务器通常可用现成 Cursor,Cline,克洛德桌面等MCP相容的应用程序.
您可以选择您想要的 OAuth 服务器提供者 , 配有 Bedrock Agent Core Gateway, 例如亚马逊·科尼托,奥克塔,或Auth0.
在您基于 stdio 的 MCP 服务器前使用 Bedrock Agent Core Gateway 需要 您可以获取 MCP 服务器的工具计划, 并在 代理 Core Gateway Lambda 目标配置。 。 。 . Agent Core Gateway随后可以给HTTP客户端做广告,验证请求输入和输出.
要获取和保存您的基于 stdio 的 MCP 服务器的工具计划到文件, 请运行 :
npx @modelcontextprotocol/inspector --cli --method tools/list <your MCP server command and arguments> > tool-schema.json
# For example:
npx @modelcontextprotocol/inspector --cli --method tools/list uvx mcp-server-time > tool-schema.json
一些MCP服务器生成 Agent Core Gateway 以严格的验证拒绝的工具计划,
例如 "items": {}, (中文(简体) ), "default": null,或 anyOf 与 {"type": "null"}。 。 。 .
在使用前你可能需要清理一下计划:
python3 scripts/clean-tool-schema.py tool-schema.json
Python 服务器实例
import sys
from mcp.client.stdio import StdioServerParameters
from mcp_lambda import BedrockAgentCoreGatewayTargetHandler, StdioServerAdapterRequestHandler
server_params = StdioServerParameters(
command=sys.executable,
args=[
"-m",
"my_mcp_server_python_module",
"--my-server-command-line-parameter",
"some_value",
],
)
request_handler = StdioServerAdapterRequestHandler(server_params)
event_handler = BedrockAgentCoreGatewayTargetHandler(request_handler)
def handler(event, context):
return event_handler.handle(event, context)
见一个可部署的完整实例 这里。 。 。 .
Typescript 服务器实例
import { Handler, Context } from "aws-lambda";
import {
BedrockAgentCoreGatewayTargetHandler,
StdioServerAdapterRequestHandler,
} from "@aws/run-mcp-servers-with-aws-lambda";
const serverParams = {
command: "npx",
args: [
"--offline",
"my-mcp-server-typescript-module",
"--my-server-command-line-parameter",
"some_value",
],
};
const requestHandler = new BedrockAgentCoreGatewayTargetHandler(
new StdioServerAdapterRequestHandler(serverParams)
);
export const handler: Handler = async (
event: Record<string, unknown>,
context: Context
): Promise<Record<string, unknown>> => {
return requestHandler.handle(event, context);
};
见一个可部署的完整实例 这里。 。 。 .
Python 客户端示例
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
# Create OAuth client provider here
async with streamablehttp_client(
url="https://abc123.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp",
auth=oauth_client_provider,
) as (
read_stream,
write_stream,
_,
):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
tool_result = await session.call_tool("echo", {"message": "hello"})
将一个完整的例子作为聊天样本的一部分 这里。 。 。 .
打印客户端示例
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
const client = new Client(
{
name: "my-client",
version: "0.0.1",
},
{
capabilities: {
sampling: {},
},
}
);
// Create OAuth client provider here
const transport = new StreamableHTTPClientTransport(
"https://abc123.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp",
{
authProvider: oauthProvider,
}
);
await client.connect(transport);
将一个完整的例子作为聊天样本的一部分 这里。 。 。 .
使用 Lambda 函数 URL
flowchart LR
App["MCP Client"]
T1["MCP Server<br>(Lambda function)"]
T2["Lambda function URL"]
App -->|"Custom Streamable HTTP<br>Transport with AWS Auth"| T2
T2 -->|"Invoke"| T1
此解决方案使用 AWS IAM 认证,并依赖于授予 Lambda Invoke FunctionUrl 许可( 允许) 给您的 IAM用户和角色可以访问MCP服务器. 客户端必须使用 MCP 扩展名 以 HTTP 签名请求的运输 AWS 西格维4。 。 。 . 与现成的MCP兼容的应用程序不太可能得到这种自定义运输的支持, 因此,这个解决方案更适合服务到服务的通信,而不是终端用户.
Python 服务器实例
import sys
from mcp.client.stdio import StdioServerParameters
from mcp_lambda import LambdaFunctionURLEventHandler, StdioServerAdapterRequestHandler
server_params = StdioServerParameters(
command=sys.executable,
args=[
"-m",
"my_mcp_server_python_module",
"--my-server-command-line-parameter",
"some_value",
],
)
request_handler = StdioServerAdapterRequestHandler(server_params)
event_handler = LambdaFunctionURLEventHandler(request_handler)
def handler(event, context):
return event_handler.handle(event, context)
见一个可部署的完整实例 这里。 。 。 .
Typescript 服务器实例
import {
Handler,
Context,
APIGatewayProxyEventV2WithIAMAuthorizer,
APIGatewayProxyResultV2,
} from "aws-lambda";
import {
LambdaFunctionURLEventHandler,
StdioServerAdapterRequestHandler,
} from "@aws/run-mcp-servers-with-aws-lambda";
const serverParams = {
command: "npx",
args: [
"--offline",
"my-mcp-server-typescript-module",
"--my-server-command-line-parameter",
"some_value",
],
};
const requestHandler = new LambdaFunctionURLEventHandler(
new StdioServerAdapterRequestHandler(serverParams)
);
export const handler: Handler = async (
event: APIGatewayProxyEventV2WithIAMAuthorizer,
context: Context
): Promise<APIGatewayProxyResultV2> => {
return requestHandler.handle(event, context);
};
见一个可部署的完整实例 这里。 。 。 .
Python 客户端示例
from mcp import ClientSession
from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client
async with aws_iam_streamablehttp_client(
endpoint="https://url-id-12345.lambda-url.us-west-2.on.aws",
aws_service="lambda",
aws_region="us-west-2",
) as (
read_stream,
write_stream,
_,
):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
tool_result = await session.call_tool("echo", {"message": "hello"})
将一个完整的例子作为聊天样本的一部分 这里。 。 。 .
打印客户端示例
import { StreamableHTTPClientWithSigV4Transport } from "@aws/run-mcp-servers-with-aws-lambda";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
const client = new Client(
{
name: "my-client",
version: "0.0.1",
},
{
capabilities: {
sampling: {},
},
}
);
const transport = new StreamableHTTPClientWithSigV4Transport(
new URL("https://url-id-12345.lambda-url.us-west-2.on.aws"),
{
service: "lambda",
region: "us-west-2",
}
);
await client.connect(transport);
将一个完整的例子作为聊天样本的一部分 这里。 。 。 .
使用 Lambda Invoke API
flowchart LR
App["MCP Client"]
T1["MCP Server<br>(Lambda function)"]
App -->|"Custom MCP Transport<br>(Lambda Invoke API)"| T1
和Lambda函数 URL方法一样,这个解决方案使用AWS IAM进行认证. 它依赖给予 Lambda Invoke功能许可 到您的 IAM 用户和角色允许访问 MCP 服务器. 客户端必须使用自定义的 MCP 运输,直接调用 Lambda Invoke API (英语)。 。 。 . 与现成的MCP兼容的应用程序不太可能得到这种自定义运输的支持, 因此,这个解决方案更适合服务到服务的通信,而不是终端用户.
Python 服务器实例
import sys
from mcp.client.stdio import StdioServerParameters
from mcp_lambda import stdio_server_adapter
server_params = StdioServerParameters(
command=sys.executable,
args=[
"-m",
"my_mcp_server_python_module",
"--my-server-command-line-parameter",
"some_value",
],
)
def handler(event, context):
return stdio_server_adapter(server_params, event, context)
见一个可部署的完整实例 这里。 。 。 .
Typescript 服务器实例
import { Handler, Context } from "aws-lambda";
import { stdioServerAdapter } from "@aws/run-mcp-servers-with-aws-lambda";
const serverParams = {
command: "npx",
args: [
"--offline",
"my-mcp-server-typescript-module",
"--my-server-command-line-parameter",
"some_value",
],
};
export const handler: Handler = async (event, context: Context) => {
return await stdioServerAdapter(serverParams, event, context);
};
见一个可部署的完整实例 这里。 。 。 .
Python 客户端示例
from mcp import ClientSession
from mcp_lambda import LambdaFunctionParameters, lambda_function_client
server_params = LambdaFunctionParameters(
function_name="my-mcp-server-function",
region_name="us-west-2",
)
async with lambda_function_client(server_params) as (
read_stream,
write_stream,
):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
tool_result = await session.call_tool("echo", {"message": "hello"})
将一个完整的例子作为聊天样本的一部分 这里。 。 。 .
打印客户端示例
import {
LambdaFunctionParameters,
LambdaFunctionClientTransport,
} from "@aws/run-mcp-servers-with-aws-lambda";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
const serverParams: LambdaFunctionParameters = {
functionName: "my-mcp-server-function",
regionName: "us-west-2",
};
const client = new Client(
{
name: "my-client",
version: "0.0.1",
},
{
capabilities: {
sampling: {},
},
}
);
const transport = new LambdaFunctionClientTransport(serverParams);
await client.connect(transport);
将一个完整的例子作为聊天样本的一部分 这里。 。 。 .
相关项目
- 在Lambda函数中写入自定义的 MCP 服务器, 见 MCP Lambda 处理器 项目.
- 通过stdio MCP服务器,将现有的Lambda功能用作工具, 见 AWS Lambda 工具 MCP 服务器 项目.
考虑因素
- 这个库目前支持用Python和Typescript编写的MCP服务器和客户端. 其他语言如克特林语不被支持.
- 这个库只为Lambda改装了stdio MCP服务器,而不是为SSE等其它协议所编写的服务器.
- 此库不维护任何 MCP 服务器状态或整个 Lambda 函数引用的会话 . 只有无国籍的MCP服务器才适合使用这个库. 例如,MCP服务器 利用无国籍工具,如 时间 MCP 服务器 或提出无国籍的网络请求,例如: 获取 MCP 服务器。 。 。 . 状态化的MCP服务器并不合适,因为每次请求都会失去状态. 例如,在磁盘或内存中管理数据的MCP服务器,例如 页:1 sqlite MCP 服务器, (中文(简体) ), 页:1 文件系统 MCP 服务器, (中文(简体) ), 编辑 git MCP 服务器。 。 。 .
部署和运行实例
见 发展指南 用于在此仓库中部署和运行示例.
警卫
见 捐款 更多信息.
许可证
这个项目根据Apache-2.0许可证获得许可证.