news 2026/9/9 21:29:14

Serverless Framework + Bedrock AgentCore:用 LangGraph Gateway 示例把 Lambda 函数封装成 Agent 工具

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
Serverless Framework + Bedrock AgentCore:用 LangGraph Gateway 示例把 Lambda 函数封装成 Agent 工具

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

一次完整请求的旅程是:

  1. 用户输入经invoke_agent_runtime到达 AgentCore Runtime;
  2. Runtime 调用智能体容器中的agent_invocation()入口(见 agent.py);
  3. LangGraph 的聊天节点让 Claude Sonnet 4.5 决定直接回答还是借助工具;
  4. 当模型判定需要计算时,LangGraph 的 ToolNode 通过 MCP 客户端把参数发往Gateway
  5. 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?——纯对话,不依赖任何工具,用于对照验证无工具路径。

注意脚本会为每次调用生成独立的runtimeSessionIduuid.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: {}

逐项拆解其含义:

  1. Lambda Function(函数)calculatorFunction是普通的 Python 3.13 Lambda,handler 指向handlers/calculator.handler
  2. Tool Definition(工具定义)ai.tools.calculator把上述函数映射为一个网关工具,toolSchema用 JSON Schema 描述工具名称、用途和入参结构。入参描述写得越具体,LLM 越容易正确生成参数;
  3. 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.pyinvoke_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仅开放pie
  • 递归遍历 AST 节点_evaluate_node,对不支持的语法直接抛出ValueError,天然规避代码注入。

同时它对两种入参形态做了兼容:既支持普通 Lambda 直接调用(event中直接取expression),也支持Gateway 转发格式(参数被包在event.body字段里,可能是字符串需要二次json.loads)。返回统一为{statusCode, body}结构,body 内含resultexpression

容器化:Dockerfile 与依赖

Dockerfile 展示了 Agent 镜像的最小形态:基于python:3.12-slim,先拷贝pyproject.tomlpip install .安装依赖,再拷贝agent.py,最后以python agent.py作为容器启动命令。

pyproject.toml 中集中列出了打通整条链路所需的依赖及其最低版本要求:

依赖最低版本作用
bedrock-agentcore1.3.0AgentCore Runtime 应用框架与入口装饰器
langgraph1.0.8状态图、ToolNode、条件边等编排能力
langchain/langchain-core1.2.10 / 1.2.13Chat 模型封装与消息类型
langchain-aws1.2.5Bedrock Converse 模型提供方
langchain-mcp-adapters0.1.0将 MCP 工具转换为 LangChain 工具
mcp1.0.0MCP 客户端会话与协议
mcp-proxy-for-aws1.1.5对 MCP 端点附加 AWS SigV4 认证

注意serverless.yml中函数 runtime 为python3.13,而基础镜像使用python:3.12-slim——两者互不冲突,前者描述的是 Lambda 函数运行时,后者是 Agent 容器自身的基础镜像。

文件清单速览

文件用途
serverless.yml基础设施配置(函数、工具、Agent)
agent.pyLangGraph Agent,含网关工具发现逻辑
handlers/calculator.pyCalculator Lambda 工具实现
DockerfileAgent 容器定义
pyproject.tomlPython 依赖声明
test-invoke.pyboto3 联调测试脚本(含三种用例)

扩展更多工具

工具不是写死在代码里的,而是在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),仅供参考

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/9/9 21:27:09

Telegraf:用一个TOML配置快速跑通指标采集全链路

Telegraf:用一个TOML配置快速跑通指标采集全链路 【免费下载链接】telegraf Agent for collecting, processing, aggregating, and writing metrics, logs, and other arbitrary data. 项目地址: https://gitcode.com/GitHub_Trending/te/telegraf 十几台服务…

作者头像 李华
网站建设 2026/9/9 21:25:58

ibaAnalyzer V6.3.1工业数据分析实操指南:从波形分析到故障诊断

简介:这是一款面向工业数据分析和IT专业人员的专业工具,专门用于解析和可视化ibapda生成的dat文件,解决海量原始数据难以理解和处理的问题。压缩包共25个文件,约40MB,核心包括Windows安装程序(.exe&#xf…

作者头像 李华
网站建设 2026/9/9 21:23:37

Pytest二次开发核心指南:从Hook机制到框架封装

做测试的人,几乎都绕不开 Pytest。但“怎么才能算对 Pytest 做二次开发”这个问题,我在社区里、面试中、团队内部都见过太多误解了。有人写了几个 fixture 封装一下接口请求,就说自己做了二次开发;有人把项目的公共方法整理到一个…

作者头像 李华
网站建设 2026/9/9 21:23:17

STM32 USART3 DMA收发详解:通道配置、发送判断与不定长接收

简介:面向嵌入式开发者的STM32F103标准库工程,专注于利用DMA方式实现USART3的接收与发送,解决传统串口中断频繁占用CPU的问题,适用于工业通信、数据采集等实时性较高的场景。压缩包共171个文件、约3.41MB,包含C源码、H…

作者头像 李华
网站建设 2026/9/9 21:22:57

ha_xiaomi_home 米家集成指南:10 分钟把小米设备搬进 Home Assistant

ha_xiaomi_home 米家集成指南:10 分钟把小米设备搬进 Home Assistant 【免费下载链接】ha_xiaomi_home Xiaomi Home Integration for Home Assistant 项目地址: https://gitcode.com/GitHub_Trending/ha/ha_xiaomi_home 你手头有小米的空气净化器、智能灯、扫…

作者头像 李华
网站建设 2026/9/9 21:20:06

Python学生成绩管理系统:入门作业的函数与数据结构实战

1. 作业内容与题目拆解1.1 这道题到底在考什么第三次Python作业,在很多入门课程里是一道分水岭。前两次作业大家还在和print、变量、if/else较劲,到了第三次,就突然要求你"组织代码"了——函数、列表、字典、字符串处理&#xff0c…

作者头像 李华