Serverless Framework + Bedrock AgentCore:用 LangGraph Gateway 示例把 Lambda 函数封装成 Agent 工具
【免费下载链接】serverless⚡ Serverless Framework – Effortlessly build apps that auto-scale, incur zero costs when idle, and require minimal maintenance using AWS Lambda and other managed cloud services.项目地址: https://gitcode.com/GitHub_Trending/se/serverless
本篇文章基于se/serverless仓库中bedrock-agentcore插件的 Python 示例langgraph-gateway,完整讲解如何将普通 Lambda 函数声明为ai.tools工具、由框架自动创建 AgentCore Gateway,并通过 MCP 协议让 LangGraph Agent 在运行时发现和调用这些工具。读完本文,你将掌握一套可直接复制的"自定义 Lambda 工具 + Claude 智能体编排"落地范式,以及配套的部署、联调与清理流程。
示例要解决的核心问题
langgraph-gateway是一个基于 LangGraph 的智能体示例,它的核心演示目标只有一个:通过 AgentCore Gateway 把自定义 Lambda 函数暴露为 Agent 可调用的工具。具体来说,这个示例展示的能力包括:
- Gateway Tools(网关工具):将 Lambda 函数暴露给智能体当作工具使用;
- Auto-Created Gateway(自动创建网关):只要定义了工具,框架会默认自动创建一个 Gateway,无需手写网关配置;
- MCP Protocol(MCP 协议):工具发现与调用统一走 Model Context Protocol;
- BedrockAgentCoreApp:与 AgentCore Runtime 集成所采用的入口模式;
- LangGraph:通过工具节点(ToolNode)编排整个智能体执行流程。
场景设定非常贴近真实业务:示例内置了一个"计算器"Lambda(calculator),Claude 收到类似 "What is 25 multiplied by 4?" 的问题时会自主决定调用计算工具,而不是自己心算,从而演示了"LLM 决策 → 网关转发 → Lambda 执行 → 结果回填"的完整闭环。
整体架构与执行链路
README 给出了清晰的调用链示意,结合示例源码可以归纳为下图所示的分层关系:
User Request → AgentCore Runtime → agent_invocation() ↓ LangGraph ↓ Claude Sonnet 4.5 ↓ [needs calculation?] ↓ Gateway (MCP) → Calculator Lambda ↓ Response一次完整请求的旅程是:
- 用户输入经
invoke_agent_runtime到达 AgentCore Runtime; - Runtime 调用智能体容器中的
agent_invocation()入口(见 agent.py); - LangGraph 的聊天节点让 Claude Sonnet 4.5 决定直接回答还是借助工具;
- 当模型判定需要计算时,LangGraph 的 ToolNode 通过 MCP 客户端把参数发往Gateway;
- Gateway 再把请求路由到 Calculator Lambda 执行,结果逐级返回给用户。
其中 Agent 与 Gateway 之间的通信走 MCP over HTTP,并借助mcp-proxy-for-aws完成 AWS SigV4 认证(详见下文源码分析)。
部署前置条件
在动手部署前需要确认以下条件齐备:
- 一个已开通Bedrock 模型访问权限的 AWS 账户(本示例使用 Claude Sonnet 4.5);
- 在 Bedrock 控制台开启美国区域推理配置文件
us.anthropic.claude-sonnet-4-5-20250929-v1:0的访问权限; - 本机已安装Docker(Agent 以容器镜像方式构建并推送至 ECR);
- Serverless Framework v4+;
- 已配置好 AWS 凭证。
快速开始:从部署到联调
1. 一键部署
在示例目录下执行:
serverless deploy根据 README 与框架的插件实现,部署过程会依次完成:
- 构建 Docker 镜像并推送到 Amazon ECR;
- 部署 Calculator Lambda 函数;
- 根据
ai.tools声明创建 AgentCore Gateway 并把 calculator 挂载为网关工具; - 部署 AgentCore Runtime;
- 输出可供调用的 Runtime ARN / URL。
关于"自动创建网关"这一步,可以从 compilation/orchestrator.js 的编译流程中找到依据:当aiConfig.gateways未显式声明时,插件会走default gateway mode——把所有共享工具集中编译到同一个默认 Gateway 上;随后在编译ai.agents下的 runtime agent 时,若该 agent 没有显式指定gateway且存在默认网关,就会自动把默认网关注入给该 agent。这正是示例配置里chatbot: {}空对象也能正常工作、README 中"Default gateway auto-created"注释的来源。
2. 发起测试调用
方式一:直接使用 boto3(对应仓库 test-invoke.py)
import boto3 import json import uuid client = boto3.client('bedrock-agentcore', region_name='us-east-1') response = client.invoke_agent_runtime( agentRuntimeArn='YOUR_RUNTIME_ARN', # 从 deploy 输出获取,也可用 serverless info 查询 runtimeSessionId=str(uuid.uuid4()), payload=json.dumps({"prompt": "What is 25 multiplied by 4?"}).encode() ) result = json.loads(response['response'].read()) print(result)方式二:直接运行示例自带测试脚本
仓库内的 test-invoke.py 封装了流式响应的处理逻辑,把工具返回的agentRuntimeArn设为环境变量后即可运行:
RUNTIME_ARN=arn:aws:bedrock-agentcore:... python3 test-invoke.py该脚本内置三个递进的测试用例,很适合用来验证"工具调用"与"普通对话"两条路径:
- Test 1:
What is 25 multiplied by 4?——触发 Gateway 上的 calculator 工具; - Test 2:
Calculate the square root of 144 plus 10——验证复杂表达式(sqrt函数)经工具正确求值; - Test 3:
Hello! What tools do you have available?——纯对话,不依赖任何工具,用于对照验证无工具路径。
注意脚本会为每次调用生成独立的runtimeSessionId(uuid.uuid4()),以便保持会话隔离。另外,示例 README 中的payload统一携带prompt键,这与 agent.py 入口处payload.get("prompt", ...)的取值逻辑一一对应。
3. 本地开发模式
如需在本地快速迭代 Agent 代码,框架提供了 dev 模式:
serverless dev配置与源码剖析:一个"计算器"工具如何跑起来
基础设施配置serverless.yml
完整的 serverless.yml 仅由三部分构成:函数声明、工具声明和 Agent 声明。
service: langgraph-gateway provider: name: aws region: us-east-1 functions: calculatorFunction: handler: handlers/calculator.handler runtime: python3.13 ai: # 定义将通过 Gateway 提供给 Agent 的工具 tools: calculator: function: calculatorFunction toolSchema: - name: calculate description: Evaluate a mathematical expression. Supports basic arithmetic (+, -, *, /, **) and functions like sqrt, sin, cos, etc. inputSchema: type: object properties: expression: type: string description: Mathematical expression to evaluate (e.g., "2 + 2 * 3", "sqrt(16)") required: - expression # LangGraph agent with gateway tools # 未显式声明网关时,框架会自动创建包含全部工具的默认网关 agents: chatbot: {}逐项拆解其含义:
- Lambda Function(函数):
calculatorFunction是普通的 Python 3.13 Lambda,handler 指向handlers/calculator.handler; - Tool Definition(工具定义):
ai.tools.calculator把上述函数映射为一个网关工具,toolSchema用 JSON Schema 描述工具名称、用途和入参结构。入参描述写得越具体,LLM 越容易正确生成参数; - Agent(智能体):
ai.agents.chatbot是一个极简 Agent,它不需要手写任何配置——Gateway URL 会由框架自动注入(见下文环境变量说明)。
Agent 侧源码:工具发现与工具调用
agent.py 完整示范了 Agent 侧三个关键环节。
第一步:从环境变量拿到网关地址
框架部署 Runtime 容器时会注入BEDROCK_AGENTCORE_GATEWAY_URL,这正是 orchestrator 编译逻辑中"默认网关注入"在运行时的落地体现:
GATEWAY_URL = os.environ.get("BEDROCK_AGENTCORE_GATEWAY_URL") AWS_REGION = os.environ.get("AWS_REGION", "us-east-1")README 中给出的会话级 MCP 工具发现代码正是它的简化版:
GATEWAY_URL = os.environ.get("BEDROCK_AGENTCORE_GATEWAY_URL") async with sse_client(GATEWAY_URL) as streams: async with ClientSession(*streams) as session: await session.initialize() tools = await session.list_tools()第二步:带 SigV4 认证的 MCP 会话
真实示例没有使用sse_client,而是通过mcp-proxy-for-aws建立带 AWS IAM SigV4 认证的流式 HTTP MCP 客户端,并把发现的工具转成 LangChain 工具:
mcp_client = aws_iam_streamablehttp_client( endpoint=GATEWAY_URL, aws_region=AWS_REGION, aws_service="bedrock-agentcore" ) async with mcp_client as (read, write, session_id_callback): async with ClientSession(read, write) as session: await session.initialize() tools = await load_mcp_tools(session) llm_with_tools = llm.bind_tools(tools)这里aws_service="bedrock-agentcore"声明了签名所用服务名,框架创建的网关会校验该签名,保证只有具备相应 IAM 权限的容器才能调用。
第三步:工具进入 LangGraph 图并被执行
当 LLM 决定使用工具时,LangGraph 的ToolNode会在图内通过 MCP 客户端发起调用:
result = await session.call_tool("calculate", {"expression": "25 * 4"})示例在图构建层面把整条回路串起来:chatbot节点用绑定了工具的 LLM 做决策,随后tools_condition判断是否需要进入ToolNode,执行完再回到chatbot:
graph_builder.add_node("chatbot", async_chatbot) if tools: graph_builder.add_node("tools", ToolNode(tools=tools)) graph_builder.add_conditional_edges("chatbot", tools_condition) graph_builder.add_edge("tools", "chatbot") graph_builder.add_edge(START, "chatbot")值得留意的是 agent 设计中的一个容错细节:若容器未收到BEDROCK_AGENTCORE_GATEWAY_URL(例如本地以裸进程方式调试),run_agent_with_gateway会打印提示并退回无工具模式,仅保留chatbot节点的纯对话图——这意味着该示例即使脱离网关也能正常回答普通问题,只是无法执行计算。此外,示例注释说明每次调用都会创建全新的 MCP 会话,以避免会话状态残留导致工具异常。
入口约定:BedrockAgentCoreApp
文件底部通过装饰器声明了 Runtime 的调用入口:
app = BedrockAgentCoreApp() @app.entrypoint async def agent_invocation(payload, context): user_message = payload.get("prompt", "Hello! How can I help you?") final_message = await run_agent_with_gateway(user_message) return {"result": final_message} if __name__ == "__main__": app.run()这里的关键约定是:Runtime 会把用户输入作为含prompt键的payload传入;函数返回结构中的result键将作为最终回答返回给调用方(test-invoke.py的invoke_agent_runtime(payload=...)正是与该约定配对)。底层依赖来自 pyproject.toml 中的bedrock-agentcore>=1.3.0。
工具侧源码:Calculator Lambda 的安全实现
作为"被网关暴露的工具",handlers/calculator.py 本身是个值得借鉴的最小工具样例,它没有用eval(),而是用 Python AST 解析实现安全的数学表达式求值:
- 运算符白名单:
OPERATORS显式映射+ - * / ** %与一元正负号; - 函数白名单:
FUNCTIONS只放行sqrt / sin / cos / tan / log / log10 / abs / floor / ceil / round / pow; - 常量白名单:
CONSTANTS仅开放pi与e; - 递归遍历 AST 节点
_evaluate_node,对不支持的语法直接抛出ValueError,天然规避代码注入。
同时它对两种入参形态做了兼容:既支持普通 Lambda 直接调用(event中直接取expression),也支持Gateway 转发格式(参数被包在event.body字段里,可能是字符串需要二次json.loads)。返回统一为{statusCode, body}结构,body 内含result与expression。
容器化:Dockerfile 与依赖
Dockerfile 展示了 Agent 镜像的最小形态:基于python:3.12-slim,先拷贝pyproject.toml并pip install .安装依赖,再拷贝agent.py,最后以python agent.py作为容器启动命令。
pyproject.toml 中集中列出了打通整条链路所需的依赖及其最低版本要求:
| 依赖 | 最低版本 | 作用 |
|---|---|---|
bedrock-agentcore | 1.3.0 | AgentCore Runtime 应用框架与入口装饰器 |
langgraph | 1.0.8 | 状态图、ToolNode、条件边等编排能力 |
langchain/langchain-core | 1.2.10 / 1.2.13 | Chat 模型封装与消息类型 |
langchain-aws | 1.2.5 | Bedrock Converse 模型提供方 |
langchain-mcp-adapters | 0.1.0 | 将 MCP 工具转换为 LangChain 工具 |
mcp | 1.0.0 | MCP 客户端会话与协议 |
mcp-proxy-for-aws | 1.1.5 | 对 MCP 端点附加 AWS SigV4 认证 |
注意serverless.yml中函数 runtime 为python3.13,而基础镜像使用python:3.12-slim——两者互不冲突,前者描述的是 Lambda 函数运行时,后者是 Agent 容器自身的基础镜像。
文件清单速览
| 文件 | 用途 |
|---|---|
| serverless.yml | 基础设施配置(函数、工具、Agent) |
| agent.py | LangGraph Agent,含网关工具发现逻辑 |
| handlers/calculator.py | Calculator Lambda 工具实现 |
| Dockerfile | Agent 容器定义 |
| pyproject.toml | Python 依赖声明 |
| test-invoke.py | boto3 联调测试脚本(含三种用例) |
扩展更多工具
工具不是写死在代码里的,而是在serverless.yml中声明式注册的,因此增加新工具只需两步:多声明一个函数 + 在ai.tools下多挂一个toolSchema。README 给出的"计算器 + 天气查询"双工具配置即为此模式:
functions: calculatorFunction: handler: handlers/calculator.handler runtime: python3.13 weatherFunction: handler: handlers/weather.handler runtime: python3.13 ai: tools: calculator: function: calculatorFunction toolSchema: [...] weather: function: weatherFunction toolSchema: - name: get_weather description: Get current weather for a city inputSchema: type: object properties: city: type: string required: - city agents: chatbot: {}从 orchestrator.js 的实现可以推断,default gateway 模式会收集ai.tools下的全部共享工具统一编译挂载,所以无需修改 Agent 端代码——新部署后 Agent 在启动时通过 MCPlist_tools就能自动发现get_weather。只要注意新 Lambda 的 handler 遵循同样的入参(含 Gatewaybody包裹)与出参格式约定即可。
清理资源
示例服务会创建容器镜像、Lambda、网关与 Runtime 等多项 AWS 资源,验证完毕后建议及时清理以免持续产生费用:
serverless remove下一步进阶方向
若希望在此基础上继续深化,仓库中提供了两个同主题的 Python 示例,可作为直接延续:
- LangGraph Multi-Gateway(多网关):在
ai.gateways下显式声明多个网关并配置不同授权,对应 orchestrator 中的多网关编译路径; - LangGraph Memory(会话记忆):在示例中为对话补充持久化能力。
若需对比 JavaScript/TypeScript 技术栈的等价实现,还可参考同目录下的examples/javascript/langgraph-gateway。
【免费下载链接】serverless⚡ Serverless Framework – Effortlessly build apps that auto-scale, incur zero costs when idle, and require minimal maintenance using AWS Lambda and other managed cloud services.项目地址: https://gitcode.com/GitHub_Trending/se/serverless
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考