1. 为什么要在 Lambda 上跑向量检索 MCP Server如果你正在给 AI Agent 接一套语义检索能力大概率会遇到三个绕不开的问题向量库要常驻、检索服务要扩容、模型调用要管 Key。传统做法是买一台 EC2 常驻跑 FastAPI前面挂 Nginx后面连 OpenSearch再自己写一套鉴权。流量低谷时机器空转烧钱流量高峰时又要手动扩容运维成本比业务代码还高。MCP Server 的出现让这件事有了新解法。它把「工具」以标准协议暴露给 Claude、Cursor、Strands Agent 这类客户端Agent 不需要知道你的检索后端是 OpenSearch 还是别的只要按 MCP 协议调用工具即可。而 Serverless 架构AWS Lambda API Gateway恰好补上了弹性这一环没有请求时不产生计算费用有请求时自动并发配合 OpenSearch 的 k-NN 向量字段就能搭出一个零运维、按量计费的语义检索服务端。这篇要交付的东西很具体一份可复制的serverless.ymlSAM 模板、MCP Server 的工具注册骨架、TaoToken 统一 Key 接入的settings.json片段以及本地调用和云端验证的完整动作。适合已经了解 MCP 基本概念、想把它落到 AWS 上的后端或 AI 应用开发者。整个链路里模型调用统一走 TaoToken 的 API 通道省去在多个厂商之间切换 Key 的麻烦。2. TaoToken 前置统一 Key 与接入通道在动手写 Lambda 之前先把模型调用这一层理顺。向量检索 MCP Server 需要两类模型能力一是把文本转成向量的 embedding 模型二是 Agent 侧对话用的对话模型。如果每个都单独申请 Key、单独配环境变量Lambda 的环境变量会越堆越多轮换时也容易漏。TaoToken 在这里的角色是统一入口。你只需要在控制台创建一个 API Key之后 embedding 请求和对话请求都走同一个 base URL 和同一个 Key。对 Lambda 来说环境变量从「N 个厂商 Key」收敛成「一个TAOTOKEN_API_KEY」代码里也不用为不同厂商写不同的请求适配。具体操作路径打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后进入控制台在 API Keys 页面新建一个 Key。建议按用途命名比如mcp-vector-lambda方便后续在 CloudWatch 日志里定位调用来源。创建完成后复制 Key它只会完整显示一次。拿到 Key 之后接入文档在 https://taotoken.net/api 可以查到完整的请求格式。embedding 接口兼容 OpenAI 的/v1/embeddings规范所以你在 Lambda 里可以直接用requests或httpx发 POST不需要额外 SDK。base URL 填https://taotoken.net/api鉴权头是Authorization: Bearer 你的Key。有一点要提醒Lambda 的环境变量里不要明文写 Key。用 SAM 的--parameter-overrides传入或者接 AWS Systems Manager Parameter Store模板里用{{resolve:ssm:...}}引用。下面第 3 节的配置会体现这一点。3. 可复制配置serverless.yml 与 MCP 工具骨架3.1 SAM 模板 serverless.yml这份模板定义了 Lambda 函数、API Gateway、DynamoDB 会话表和必要的 IAM 权限。OpenSearch 的 endpoint 和 TaoToken 的 Key 都通过参数传入不写死在文件里。AWSTemplateFormatVersion: 2010-09-09 Transform: AWS::Serverless-2016-10-31 Description: Vector Search MCP Server on Lambda with OpenSearch Parameters: McpAuthToken: Type: String NoEcho: true OpenSearchHost: Type: String OpenSearchUsername: Type: String OpenSearchPassword: Type: String NoEcho: true TaoTokenApiKey: Type: String NoEcho: true Globals: Function: Timeout: 30 MemorySize: 512 Runtime: python3.11 Environment: Variables: OPENSEARCH_HOST: !Ref OpenSearchHost OPENSEARCH_USERNAME: !Ref OpenSearchUsername OPENSEARCH_PASSWORD: !Ref OpenSearchPassword TAOTOKEN_API_KEY: !Ref TaoTokenApiKey TAOTOKEN_BASE_URL: https://taotoken.net/api MCP_AUTH_TOKEN: !Ref McpAuthToken SESSION_TABLE: !Ref SessionTable Resources: McpFunction: Type: AWS::Serverless::Function Properties: CodeUri: src/ Handler: app.lambda_handler Events: McpApi: Type: Api Properties: Path: /mcp Method: post RestApiId: !Ref McpApiGateway Policies: - DynamoDBCrudPolicy: TableName: !Ref SessionTable McpApiGateway: Type: AWS::Serverless::Api Properties: StageName: prod Auth: DefaultAuthorizer: McpTokenAuthorizer Authorizers: McpTokenAuthorizer: FunctionArn: !GetAtt AuthFunction.Arn Identity: Header: authorizationToken AuthFunction: Type: AWS::Serverless::Function Properties: CodeUri: src/ Handler: auth.lambda_handler Runtime: python3.11 SessionTable: Type: AWS::DynamoDB::Table Properties: TableName: mcp-session-table BillingMode: PAY_PER_REQUEST AttributeDefinitions: - AttributeName: session_id AttributeType: S KeySchema: - AttributeName: session_id KeyType: HASH TimeToLiveSpecification: AttributeName: ttl Enabled: true Outputs: McpEndpoint: Description: API Gateway endpoint for MCP Server Value: !Sub https://${McpApiGateway}.execute-api.${AWS::Region}.amazonaws.com/prod/mcp几个关键点NoEcho: true保证敏感参数不会在 CloudFormation 控制台回显DynamoDB 开了 TTL会话过期自动清理不用写定时任务API Gateway 挂了自定义授权器所有请求先过AuthFunction校验 token。3.2 MCP 工具注册骨架Lambda 里的 MCP Server 核心是一个装饰器注册机制。下面这份骨架把「文本索引」和「相似度检索」两个工具注册进去embedding 调用统一走 TaoToken。import json import os import requests from typing import Dict TAOTOKEN_BASE_URL os.environ[TAOTOKEN_BASE_URL] TAOTOKEN_API_KEY os.environ[TAOTOKEN_API_KEY] EMBEDDING_MODEL BAAI/bge-m3 def generate_embedding(text: str) - Dict: 通过 TaoToken 统一通道生成文本向量 url f{TAOTOKEN_BASE_URL}/v1/embeddings headers { Authorization: fBearer {TAOTOKEN_API_KEY}, Content-Type: application/json, } payload { model: EMBEDDING_MODEL, input: text, encoding_format: float, } try: resp requests.post(url, jsonpayload, headersheaders, timeout15) resp.raise_for_status() data resp.json() return { status: success, embedding: data[data][0][embedding], model: EMBEDDING_MODEL, } except Exception as e: return {status: error, message: str(e)} class LambdaMCPServer: def __init__(self): self.tools {} def tool(self): def decorator(func): self.tools[func.__name__] func return func return decorator mcp_server LambdaMCPServer() mcp_server.tool() def index_text_with_embedding(text: str, document_id: str None, metadata: str {}) - Dict: 将文本转换为向量并索引到 OpenSearch 知识库 emb generate_embedding(text) if emb[status] ! success: return emb # 此处调用 OpenSearchClient.write_document 写入 knn_vector 字段 return {status: success, document_id: document_id} mcp_server.tool() def text_similarity_search(text: str, k: int 10, score: float 0.0) - Dict: 基于向量相似度检索相关文档 emb generate_embedding(text) if emb[status] ! success: return emb # 此处调用 OpenSearchClient.search_documents 执行 k-NN 查询 return {status: success, query_vector_dim: len(emb[embedding])}generate_embedding里只认TAOTOKEN_BASE_URL和TAOTOKEN_API_KEY两个环境变量换模型只改EMBEDDING_MODEL常量不用动请求逻辑。这就是统一 Key 通道带来的直接好处。3.3 TaoToken 接入 settings.json 片段如果你在本地用 Claude Code 或 Cursor 这类支持 MCP 的客户端调试需要在settings.json里声明 MCP Server 和模型通道。下面这段同时配了 TaoToken 的对话通道和本地 MCP Server。{ mcpServers: { vector-search: { url: https://your-api-id.execute-api.cn-north-1.amazonaws.com/prod/mcp, headers: { authorizationToken: your-mcp-auth-token } } }, modelProviders: { taotoken: { baseUrl: https://taotoken.net/api, apiKey: sk-your-taotoken-key, models: { chat: claude-sonnet-4-5, embedding: BAAI/bge-m3 } } } }mcpServers里的url填第 3.1 节 SAM 部署后输出的McpEndpointauthorizationToken填你传给McpAuthToken参数的值。modelProviders里的 Key 就是第 2 节在控制台创建的那个。这样本地客户端既能调云端 MCP 工具又能通过 TaoToken 走对话模型两边共用一个 Key。4. 验证请求与成功结果4.1 本地先验证 embedding 通道在部署 Lambda 之前先用 curl 确认 TaoToken 的 embedding 接口通。这一步能排除掉 90% 的鉴权问题。curl -X POST https://taotoken.net/api/v1/embeddings \ -H Authorization: Bearer sk-your-taotoken-key \ -H Content-Type: application/json \ -d { model: BAAI/bge-m3, input: 厄尔尼诺监测系统, encoding_format: float }返回体里data[0].embedding是一个长度 1024 的浮点数组。如果返回 401检查 Key 是否复制完整如果返回 404检查 base URL 有没有多写或少写/v1。4.2 部署并验证 MCP 工具列表用 SAM 部署参数通过命令行传入不落盘sam build sam deploy --guided \ --parameter-overrides \ McpAuthTokenyour-mcp-token \ OpenSearchHostyour-opensearch-endpoint \ OpenSearchUsernameadmin \ OpenSearchPasswordyour-password \ TaoTokenApiKeysk-your-taotoken-key部署完成后用 MCP 客户端连接McpEndpoint。连接成功后点「List Tools」应该能看到index_text_with_embedding和text_similarity_search两个工具。选中text_similarity_search传入{text: 厄尔尼诺监测系统, k: 5, score: 0.5}正常返回里会带query_vector_dim: 1024说明 embedding 通道和工具注册都通了。4.3 用 Strands Agent 调用如果你用 Strands Agent 做编排工具定义和调用可以这样写import os from strands import Agent agent Agent(tools[similarity_search.py]) API_ENDPOINT os.getenv(MCP_ENDPOINT) AUTH_TOKEN os.getenv(MCP_AUTH_TOKEN) results agent.tool.similarity_search( text厄尔尼诺监测系统, k5, score0.5, api_endpointAPI_ENDPOINT, auth_tokenAUTH_TOKEN, ) print(results)similarity_search.py里按 Strands 的 Tool 规范声明名称、描述、输入输出内部用requests发 POST 到 MCP endpoint带上authorizationToken头。返回的results就是 OpenSearch 的 k-NN 检索结果。5. 本篇常见错排查Lambda 超时 30 秒embedding 请求默认超时 15 秒如果 OpenSearch 写入慢两个加起来容易顶到 Lambda 上限。把Timeout调到 60或者把 embedding 和写入拆成两个异步步骤。API Gateway 返回 403自定义授权器里event.get(authorizationToken)取的是请求头注意大小写。API Gateway 会把头名转成小写所以客户端传authorizationToken时授权器里要用event[headers][authorizationtoken]取。OpenSearch k-NN 查询报维度不匹配索引创建时knn_vector的dimension必须和 embedding 输出维度一致。BGE-M3 是 1024 维如果你换了模型索引要重建。DynamoDB 会话读不到检查 Lambda 执行角色的 IAM 策略有没有dynamodb:GetItem和dynamodb:PutItem。SAM 模板里用的DynamoDBCrudPolicy已经覆盖但如果你手动改了策略容易漏。TaoToken 返回 429并发高了触发限流。Lambda 的预留并发调低一点或者在generate_embedding里加指数退避重试。6. 下一步把 Key 和通道固定下来整套链路跑通之后你会发现最值得固定下来的不是 Lambda 代码而是模型调用通道。Lambda 可以随时重新部署OpenSearch 索引可以重建但 Key 一旦散落在多个环境变量里轮换和审计就会变成负担。我的做法是把 TaoToken 的 Key 统一放在一个环境变量里embedding 和对话共用。本地调试用settings.json里的modelProviders云端 Lambda 用 SAM 参数注入两边指向同一个 base URL。这样无论你后面加多少个 MCP 工具、换多少个模型接入层都不用动。如果你还没建 Key可以从控制台开始https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。建完之后把 Key 填进上面的serverless.yml参数和settings.json重新sam deploy一次整条 Serverless 向量检索链路就活了。