AI导哈

MCP服务器Lambda运行器

该项目支持在AWS Lambda函数中运行基于stdio的Model Context Protocol服务器，实现分布式系统的MCP协议通信。

  • by awslabs
  • Python
  • 本地部署
  • 381 星标
  • 更新于
在 GitHub 查看 awslabs/run-model-context-protocol-servers-with-aws-lambda

使用说明

使用AWS Lambda运行模型上下文协议（MCP）服务器

此项目使您能够运行 模型上下文协议 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函数调用将：

  1. 将stdio MCP服务器作为子进程启动
  2. 初始化MCP服务器
  3. 将传入请求转发到本地服务器
  4. 将服务器的响应返回给函数调用者
  5. 关闭MCP服务器子进程

此库支持通过四种方式连接到基于Lambda的MCP服务器：

  1. 这 MCP可流式HTTP传输，使用亚马逊API网关。通常使用OAuth进行身份验证。
  2. 使用Amazon Bedrock AgentCore网关的MCP流式HTTP传输。使用OAuth进行身份验证。
  3. 使用Lambda函数URL支持SigV4的自定义流式HTTP传输。通过AWS IAM进行身份验证。
  4. 直接使用Lambda Invoke API的自定义Lambda调用传输。已通过AWS IAM身份验证。

确定服务器参数

许多基于stdio的MCP服务器的文档鼓励使用按需下载和运行服务器的工具。 例如， uvx my-mcp-server 或 npx my-mcp-server. 这些工具通常没有在Lambda环境中预先打包，并且可能效率低下 在每次Lambda调用时重新下载服务器。

相反，此存储库中的示例显示了如何将MCP服务器与 Lambda函数代码，然后以 python 或 node （或 npx --offline)直接。

您需要根据MCP服务器的软件包确定正确的参数。 这通常是一个本地试错过程，因为MCP服务器的打包方式各不相同。

Python server examples

基本示例：

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 server examples

基本示例：

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服务器 和那个 Brave搜索MCP服务器 要求API密钥向第三方API发出请求。 您可以将这些API密钥配置为 加密环境变量 在Lambda函数的配置中，或者从Lambda函数代码中的Secrets Manager中检索它们（以下示例）。 但是，请注意，任何有权调用Lambda函数的人 然后将有权使用您的API键通过调用该函数来调用第三方API。 我们建议使用以下方式限制对Lambda函数的访问 最低权限IAM策略. 如果使用基于身份的身份验证机制（如OAuth），还可以为每个用户存储和检索API密钥，但此存储库中没有实现示例。

Python server example retrieving an API key from Secrets Manager 
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 server example retrieving an API key from Secrets Manager 
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 API。 我们建议使用以下方式限制对Lambda函数的访问 最低权限IAM策略.

Python server example using AWS credentials via environment variables 
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 "",
    },
)
Python server example using AWS credentials via credentials file

一些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服务器通常可以与现成的服务器一起使用 MCP兼容的应用程序，如Cursor、Cline、Claude Desktop等。

您可以为此解决方案选择所需的OAuth服务器提供商。这里的例子 存储库使用Amazon Cognito，或者您可以使用Okta或Auth0等第三方提供商 API网关自定义授权。

Python server example 
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 server example 
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 client example 
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"})

查看示例聊天机器人中的完整示例 这里.

Typescript client example 
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);

查看示例聊天机器人中的完整示例 这里.

使用基岩试剂核心网关

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服务器通常可以与现成的服务器一起使用 MCP兼容的应用程序，如Cursor、Cline、Claude Desktop等。

您可以使用Bedrock AgentCore Gateway选择所需的OAuth服务器提供商， 例如亚马逊Cognito、Okta或Auth0。

在基于stdio的MCP服务器前使用Bedrock AgentCore网关需要 您检索MCP服务器的工具架构，并在 AgentCore网关Lambda目标配置. 然后，AgentCore 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服务器生成工具模式，AgentCore Gateway通过严格验证拒绝这些模式， 例如 "items": {}, "default": null，或 anyOf 和 {"type": "null"}. 在使用之前，您可能需要清理模式：

python3 scripts/clean-tool-schema.py tool-schema.json
Python server example 
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 server example 
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 client example 
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"})

查看示例聊天机器人中的完整示例 这里.

Typescript client example 
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调用函数URL权限 到你的 IAM用户和角色，以启用对MCP服务器的访问。客户端必须使用MCP Streamable的扩展 对请求进行签名的HTTP传输 AWS SigV4. 现成的MCP兼容应用不太可能支持这种定制传输， 因此，这种解决方案更适合服务到服务的通信，而不是最终用户。

Python server example 
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 server example 
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 client example 
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"})

查看示例聊天机器人中的完整示例 这里.

Typescript client example 
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调用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调用函数权限 您的IAM用户和角色可以访问MCP服务器。 客户端必须使用直接调用的自定义MCP传输 Lambda调用API. 现成的MCP兼容应用不太可能支持这种定制传输， 因此，这种解决方案更适合服务到服务的通信，而不是最终用户。

Python server example 
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 server example 
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 client example 
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"})

查看示例聊天机器人中的完整示例 这里.

Typescript client example 
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);

查看示例聊天机器人中的完整示例 这里.

相关项目

注意事项

  • 该库目前支持用Python和Typescript编写的MCP服务器和客户端。 不支持Kotlin等其他语言。
  • 此库仅适用于Lambda的stdio MCP服务器，而不适用于为SSE等其他协议编写的服务器。
  • 此库不维护任何MCP服务器状态或跨Lambda函数调用的会话。 只有无状态MCP服务器才适合使用此库。例如，MCP服务器 调用无状态工具，如 时间MCP服务器 或者发出无状态的web请求，如 获取MCP服务器. 有状态的MCP服务器并不适合，因为它们在每次请求时都会失去状态。 例如，管理磁盘或内存中数据的MCP服务器，如 这 sqlite MCP服务器, 这 文件系统MCP服务器, 和那个 git MCP服务器.

部署并运行示例

请参阅 开发指南 有关部署和运行此存储库中的示例的说明。

安全

看 贡献 了解更多信息。

许可证

此项目根据Apache-2.0许可证获得许可。

相关 MCP