STARWARCLOUD STARWAR.AI
返回市场
运行模型上下文协议服务器与AWS Lambda

运行模型上下文协议服务器与AWS Lambda

作者：awslabs332 星标更新：2025-11-19
安装使用来源：spider==github.com

项目介绍

技术文档摘要

在 AWS Lambda 中运行 Model Context Protocol (MCP) 服务器

PyPI - 下载量 NPM 下载量

该项目使您能够在 AWS Lambda 函数中运行基于标准 I/O 的 Model Context Protocol（MCP）服务器。

目前，大多数 MCP 服务器和客户端的实现都是在单台机器上完全本地化的。桌面应用程序如 IDE 或 Claude Desktop 会作为子进程启动 MCP 服务器，并通过长时间运行的标准 I/O 流与这些服务器进行通信。

flowchart LR
    subgraph "您的笔记本电脑"
        Host["桌面应用<br>带有 MCP 客户端"]
        S1["MCP 服务器 A<br>(子进程)"]
        S2["MCP 服务器 B<br>(子进程)"]
        Host <-->|"MCP 协议<br>(通过标准 I/O 流)"| S1
        Host <-->|"MCP 协议<br>(通过标准 I/O 流)"| S2
    end

此库帮助您将现有的基于标准 I/O 的 MCP 服务器封装到 Lambda 函数中。您可以使用 MCP 协议从您的应用程序调用这些基于函数的 MCP 服务器，通过短暂的 HTTPS 连接。您的应用程序可以是桌面应用、云中的分布式系统或任何其他架构。

flowchart LR
    subgraph "分布式系统"
        App["您的应用<br>带有 MCP 客户端"]
        S3["MCP 服务器 A<br>(Lambda 函数)"]
        S4["MCP 服务器 B<br>(Lambda 函数)"]
        App <-->|"MCP 协议<br>(通过 HTTPS 连接)"| S3
        App <-->|"MCP 协议<br>(通过 HTTPS 连接)"| S4
    end

使用此库时，Lambda 函数将管理您的标准 I/O MCP 服务器的生命周期。每次 Lambda 函数调用都会：

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

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

  1. 使用 Amazon API Gateway 的 MCP 可流式传输 HTTP 传输，通常使用 OAuth 认证。
  2. 使用 Amazon Bedrock AgentCore Gateway 的 MCP 可流式传输 HTTP 传输，使用 OAuth 认证。
  3. 使用支持 SigV4 的自定义可流式传输 HTTP 传输，使用 Lambda 函数 URL，使用 AWS IAM 认证。
  4. 使用直接调用 Lambda 调用 API 的自定义 Lambda 调用传输，使用 AWS IAM 认证。

确定您的服务器参数

许多基于标准 I/O 的 MCP 服务器的文档鼓励使用工具来按需下载并运行服务器。例如，uvx my-mcp-server 或 npx my-mcp-server。这些工具通常不在 Lambda 环境中预打包，而且在每次 Lambda 调用时重新下载服务器可能会很低效。

相反，本仓库中的示例展示了如何将 MCP 服务器与 Lambda 函数代码一起打包，然后直接使用 python 或 node（或 npx --offline）启动它。

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

<details> <summary><b>Python 服务器示例</b></summary>

基本示例：

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 # 注意子模块

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,
)
</details> <details> <summary><b>Typescript 服务器示例</b></summary>

基本示例：

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
</details>

使用 API Gateway

flowchart LR
    App["MCP 客户端"]
    T1["MCP 服务器<br>(Lambda 函数)"]
    T2["API Gateway"]
    T3["OAuth 服务器<br>(Cognito 或类似服务)"]
    App -->|"MCP 可流式传输<br>HTTP 传输"| T2
    T2 -->|"调用"| T1
    T2 -->|"授权"| T3

此解决方案兼容大多数支持可流式传输 HTTP 传输的 MCP 客户端。使用这种架构部署的 MCP 服务器通常可以与现成的 MCP 兼容应用程序（如 Cursor、Cline、Claude Desktop 等）一起使用。

您可以选择所需的 OAuth 服务器提供商。本仓库中的示例使用了 Amazon Cognito，或者您可以使用第三方提供商（如 Okta 或 Auth0）与 API Gateway 自定义授权。

<details> <summary><b>Python 服务器示例</b></summary> 
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)

查看完整的可部署示例 这里。

</details> <details> <summary><b>Typescript 服务器示例</b></summary> 
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);
};

查看完整的可部署示例 这里。

</details> <details> <summary><b>Python 客户端示例</b></summary> 
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

# 在此处创建 OAuth 客户端提供程序

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"})

查看完整的示例作为样本聊天机器人的一部分 这里。

</details> <details> <summary><b>Typescript 客户端示例</b></summary> 
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: {},
    },
  }
);

// 在此处创建 OAuth 客户端提供程序

const transport = new StreamableHTTPClientTransport(
  "https://abc123.execute-api.us-west-2.amazonaws.com/prod/mcp",
  {
    authProvider: oauthProvider,
  }
);
await client.connect(transport);

查看完整的示例作为样本聊天机器人的一部分 这里。

</details>

使用 Bedrock AgentCore Gateway

flowchart LR
    App["MCP 客户端"]
    T1["MCP 服务器<br>(Lambda 函数)"]
    T2["Bedrock AgentCore Gateway"]
    T3["OAuth 服务器<br>(Cognito 或类似服务)"]
    App -->|"MCP 可流式传输<br>HTTP 传输"| T2
    T2 -->|"调用"| T1
    T2 -->|"授权"| T3

此解决方案兼容大多数支持可流式传输 HTTP 传输的 MCP 客户端。使用这种架构部署的 MCP 服务器通常可以与现成的 MCP 兼容应用程序（如 Cursor、Cline、Claude Desktop 等）一起使用。

您可以选择所需的 OAuth 服务器提供商，例如 Amazon Cognito、Okta 或 Auth0。

在基于标准 I/O 的 MCP 服务器前使用 Bedrock AgentCore Gateway 需要检索 MCP 服务器的工具模式，并在 AgentCore Gateway Lambda 目标配置 中提供。AgentCore Gateway 可以向 HTTP 客户端宣传该模式，并验证请求的输入和输出。

要检索并保存基于标准 I/O 的 MCP 服务器的工具模式到文件，请运行：

npx @modelcontextprotocol/inspector --cli --method tools/list <您的 MCP 服务器命令和参数> > tool-schema.json

# 例如：
npx @modelcontextprotocol/inspector --cli --method tools/list uvx mcp-server-time > tool-schema.json
<details> <summary><b>Python 服务器示例</b></summary> 
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)

查看完整的可部署示例 这里。

</details> <details> <summary><b>Typescript 服务器示例</b></summary> 
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);
};

查看完整的可部署示例 这里。

</details> <details> <summary><b>Python 客户端示例</b></summary> 
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

# 在此处创建 OAuth 客户端提供程序

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"})

查看完整的示例作为样本聊天机器人的一部分 这里。

</details> <details> <summary><b>Typescript 客户端示例</b></summary> 
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: {},
    },
  }
);

// 在此处创建 OAuth 客户端提供程序

const transport = new StreamableHTTPClientTransport(
  "https://abc123.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp",
  {
    authProvider: oauthProvider,
  }
);
await client.connect(transport);

查看完整的示例作为样本聊天机器人的一部分 这里。

</details>

使用 Lambda 函数 URL

flowchart LR
    App["MCP 客户端"]
    T1["MCP 服务器<br>(Lambda 函数)"]
    T2["Lambda 函数 URL"]
    App -->|"自定义可流式传输 HTTP<br>传输带 AWS 认证"| T2
    T2 -->|"调用"| T1

此解决方案使用 AWS IAM 进行认证，并依赖于授予 Lambda 用户和角色 Lambda InvokeFunctionUrl 权限 来访问 MCP 服务器。客户端必须使用扩展的 MCP 可流式传输 HTTP 传输，该传输使用 AWS SigV4 对请求进行签名。现成的 MCP 兼容应用程序不太可能支持此自定义传输，因此此解决方案更适合服务间通信而不是最终用户。

<details> <summary><b>Python 服务器示例</b></summary> 
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)

查看完整的可部署示例 这里。

</details> <details> <summary><b>Typescript 服务器示例</b></summary> 
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);
};

查看完整的可部署示例 这里。

</details> <details> <summary><b>Python 客户端示例</b></summary> 
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 =