AI - 用 FastAPI 暴露你的第一个 Google ADK Agent

在上一篇《 AI - 使用 Google ADK 创建你的第一个 AI Agent》我们已经用 Google ADK（Agent Development Kit）写了一个最简单的 Agent：会用 Gemini 模型，能调用一个工具函数，回答“某个城市现在几点”。

这一篇，我们来做一件现实世界里更有用的事情：

把这个 Agent 用 FastAPI 暴露成 HTTP API，
支持多用户、多会话（user_id + session_id），
让前端、移动端、其他服务都可以通过 REST 调用它。

过程中会顺带拆一下几个关键技术点：

• FastAPI + Uvicorn 是怎么跑起来的
• ADK 里的 SessionService / Runner 是干嘛的
• user_id + session_id 在 ADK 中是怎么维护对话上下文的
• 异步 run_async 和流式事件到底是什么

1. 起点：我们已经有了一个最简单的 Agent

先快速回顾一下我们的 agent.py

# my_agent/agent.pyfromgoogle.adk.agents.llm_agentimportAgent# 一个非常简单的工具函数：返回某个城市的时间（写死）defget_current_time(city:str)->dict:"""Returns the current time in a specified city."""return{"status":"success","city":city,"time":"10:30 AM"}# 根 Agent（ADK 约定必须叫 root_agent）root_agent=Agent(model='gemini-3-pro-preview',# 或其他 Gemini 模型name='root_agent',description="Tells the current time in a specified city.",instruction="You are a helpful assistant that tells the current time in cities. Use the 'get_current_time' tool for this purpose.",tools=[get_current_time],)

项目结构大概是这样:

my_agent/ agent.py # main agent code __init__.py .env # API keys or project IDs

现在，它可以用 adk run my_agent 或 adk web 跑起来，但只能在本机 CLI / Dev UI 里玩，对外没有 HTTP 接口。接下来我们就基于这个项目，加一个 api.py，用 FastAPI 暴露出来。

2. 目标架构：把 Agent 变成一个 HTTP 服务

我们想要的是这样一套能力：

• POST /session：传入 user_id，创建一个新的对话会话，返回 session_id
• POST /chat：传入 user_id + session_id + message，Agent 在对应会话上继续对话并返回回复

整体逻辑：

前端 / 其他服务 │ ├─ POST /session -> 返回 session_id │ └─ POST /chat(user_id, session_id, message) │ ▼ FastAPI 路由 │ ▼ ADK Runner + SessionService (负责会话 & state) │ ▼ root_agent.run_async(...) │ ▼ Gemini + tools

其中：

• FastAPI：Python 高性能 Web 框架，非常适合写 JSON API。
• Uvicorn：ASGI 服务器，用来实际跑 FastAPI 应用。
• SessionService：ADK 中管理会话（Session）的组件，提供创建 / 获取 / 列表 / 删除等能力。
• Runner：ADK 的“执行引擎”，负责拿到 Session、调用 Agent、处理生成的 Event 并写回 Session。

3. 新增一个 api.py：FastAPI 入口 & ADK 运行时

我们计划把 api.py 放到 my_agent/ 包里，所以最终目录变成这样：

my_agent/ ├─ agent.py # 定义 root_agent ├─ api.py # ✅ 新增：FastAPI + Runner + Session └─ __init__.py

3.1 安装 FastAPI + Uvicorn + python-dotenv（可选）

在你的虚拟环境里安装：

pipinstallfastapi uvicorn[standard]python-dotenv

• FastAPI：写 API 路由，用 Pydantic 定义请求 / 响应模型
• Uvicorn：跑 ASGI 服务器
• python-dotenv：方便从 .env 自动加载 GOOGLE_API_KEY 等环境变量

3.2 会话相关：InMemorySessionService + Runner

在 ADK 里，一个 Session 就代表一段连续的对话，里面包含：

• id：session_id
• user_id：对应用户
• events：所有历史对话 / tool 调用等事件
• state：当前会话的状态字典（短期记忆）

Session 的生命周期由 SessionService 管理，例如：

• InMemorySessionService：存在内存里，应用重启就没了，很适合本地开发 / demo
• DatabaseSessionService：持久化到数据库，适合生产环境
• VertexAiSessionService：把 Session 存到 Vertex AI 的 Agent Engine 里

而 Runner 则负责：

• 根据 user_id + session_id 找到对应 Session
• 把新一轮 new_message 加到 Session 的事件里
• 调用 root_agent.run_async(…)，接收流式 Event 并处理
• 把新的事件和状态写回 Session

4. 完整的 api.py 源码（带 session 支持）

下面是一份可以直接用的 my_agent/api.py，你可以一次性拷贝过去：

# my_agent/api.py""" 用 FastAPI 暴露基于 Google ADK 的第一个 Agent。 - /session : 创建会话，返回 session_id - /chat : 带 user_id + session_id 发消息，支持多轮对话 """importosimportuvicornfromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelfromdotenvimportload_dotenvfromgoogle.adk.sessionsimportInMemorySessionServicefromgoogle.adk.runnersimportRunnerfromgoogle.genaiimporttypesasgenai_types# 导入我们在 agent.py 里定义好的 root_agentfrom.agentimportroot_agent# ===================== 环境变量 =====================# 加载同级或上级目录下的 .env（包含 GOOGLE_API_KEY 等）load_dotenv()ifnotos.getenv("GOOGLE_API_KEY"):# 不强制报错，只打印一个友好提示print("[WARN] GOOGLE_API_KEY 未设置，""请在 .env 或环境变量中配置，否则模型调用会失败。")# ===================== ADK：SessionService + Runner =====================# 应用名：ADK 用它来区分不同应用的会话空间APP_NAME="my_first_adk_agent"# 1) 会话服务：这里用内存版，适合本地开发 / demosession_service=InMemorySessionService()# 2) Runner：负责从 Session 取历史、调用 Agent、写回事件runner=Runner(app_name=APP_NAME,agent=root_agent,session_service=session_service,)# ===================== FastAPI 初始化 =====================app=FastAPI(title="My First ADK Agent API",description="基于 Google ADK + FastAPI 的简单对话 Agent，支持 user_id + session_id 会话。",)# ===================== 请求 / 响应模型 =====================classCreateSessionRequest(BaseModel):"""创建会话请求：只需要 user_id。"""user_id:strclassCreateSessionResponse(BaseModel):user_id:strsession_id:strclassChatRequest(BaseModel):"""对话请求：必须带 user_id + session_id + message。"""user_id:strsession_id:strmessage:strclassChatResponse(BaseModel):user_id:strsession_id:strreply:str# ===================== 路由：创建会话 =====================@app.post("/session",response_model=CreateSessionResponse)asyncdefcreate_session(req:CreateSessionRequest):""" 创建一个新的 Session： - 输入: user_id - 输出: 自动生成的 session_id """# 注意：SessionService 在新版本 ADK 里是异步的，需要 await。session=awaitsession_service.create_session(app_name=APP_NAME,user_id=req.user_id,state={},# 初始 state，你也可以在这里塞一些默认值)returnCreateSessionResponse(user_id=req.user_id,session_id=session.id,)# ===================== 路由：在指定会话里聊天 =====================@app.post("/chat",response_model=ChatResponse)asyncdefchat(req:ChatRequest):""" 使用 user_id + session_id + message 调用 Runner， 在对应的 Session 上继续对话并返回模型回复。 """# 把用户输入包装成 google-genai 的 Content/Part 结构，# Runner.run_async 需要的就是这个类型。user_content=genai_types.Content(role="user",parts=[genai_types.Part.from_text(text=req.message)],)reply_chunks:list[str]=[]try:# Runner.run_async 是一个异步生成器，会不断 yield Event。asyncforeventinrunner.run_async(user_id=req.user_id,session_id=req.session_id,new_message=user_content,):# 把事件中的文本部分收集起来ifevent.contentandevent.content.parts:forpartinevent.content.parts:ifgetattr(part,"text",None):reply_chunks.append(part.text)# is_final_response() 为 True 时，说明这是最终回复，可以结束本轮ifevent.is_final_response():breakexceptExceptionase:msg=str(e)# 一个常见错误是 session_id 不存在if"session"inmsg.lower()and"not found"inmsg.lower():raiseHTTPException(status_code=404,detail="会话不存在，请先调用 /session 创建会话。",)raiseHTTPException(status_code=500,detail=f"Agent 调用失败:{msg}",)reply_text="".join(reply_chunks).strip()or"（Agent 没有返回任何文本内容）"returnChatResponse(user_id=req.user_id,session_id=req.session_id,reply=reply_text,)# ===================== 本地启动（可选） =====================if__name__=="__main__":# 在项目根目录运行：python -m my_agent.apiuvicorn.run("my_agent.api:app",host="0.0.0.0",port=8001,reload=True)

5. 逐段拆解一下关键技术点

5.1 SessionService：为什么要 user_id + session_id？

ADK 把一段连续对话抽象成一个 Session 对象，由 SessionService 统一管理：

session=awaitsession_service.create_session(app_name=APP_NAME,user_id=req.user_id,state={})

• app_name：区分是哪个应用的会话（你完全可以在一个 Python 进程里跑多个 Agent）。
• user_id：谁的会话。
• session.id：ADK 生成的唯一 session_id。

在后续每一轮对话里，我们都要显式带上这两个维度：

asyncforeventinrunner.run_async(user_id=req.user_id,session_id=req.session_id,new_message=user_content,):...

这相当于告诉 Runner：

“请在 APP_NAME 这个应用里，
找到 user_id = X、session_id = Y 的那条会话，
然后在它的历史基础上继续处理这条新消息。”

ADK 会帮你自动：

• 从 SessionService 读出该 Session 的 events 和 state
• 把新消息加入事件链
• 把 Agent 产生的新事件 / state 变化写回该 Session

所以：

• user_id 让你可以区分不同用户
• session_id 让你可以区分同一用户的多段对话（多窗口 / 多任务）

如果以后要做“跨会话的长期记忆”，可以再加上 MemoryService（比如 InMemoryMemoryService / VertexAiMemoryBankService），那就是另外一层了。

5.2 Runner：为什么不用直接 root_agent.run()？

理论上你也可以在 FastAPI 里直接调用：

awaitroot_agent.run(...)

但官方更推荐用 Runner 来跑 Agent，因为 Runner 额外做了很多底层活：

• 创建 / 获取 Session
• 把新消息写入 Session 的事件链
• 处理 Agent 生成的每个 Event，包括：
  • 工具调用
  • 状态变化（state_delta）
  • 产出的内容（content）
• 把这些变化提交给 SessionService、MemoryService 等

也就是说：

Runner = Agent 的“执行引擎 + 会话协调器”

对你来说，Runner 有两个重要特性：

• run_async(…) 是异步的（适合 FastAPI 等 async 框架）
• 返回的是一个 异步生成器（AsyncGenerator[Event]），可以边生成边消费，天然契合“流式输出”的场景

在我们的 /chat 接口里，就是用一个 async for 来消费这些 Event：

asyncforeventinrunner.run_async(...):ifevent.contentandevent.content.parts:...ifevent.is_final_response():break

现在我们只是把所有文本 part 拼起来，做一个简单的“整段回复”。
但你以后完全可以玩得更高级：

• 逐块推给前端做“打字机效果”
• 根据不同类型的 Event 做不同 UI（例如工具调用 / 代码执行）

5.3 FastAPI：Pydantic 模型 + JSON Body

FastAPI 的一个特点是：用 Pydantic 模型来声明请求体 / 响应体，自动帮你做验证和文档生成。
比如我们定义：

classChatRequest(BaseModel):user_id:strsession_id:strmessage:str

然后在路由里写：

@app.post("/chat",response_model=ChatResponse)asyncdefchat(req:ChatRequest):...

FastAPI 会自动：

• 把 JSON 请求体解析成 ChatRequest 对象
• 在 /docs 里生成 Swagger UI 文档
• 检查字段类型，不符合直接 422 返回

这一点和 ADK 本身的“结构化输出”（output_schema）非常契合——
前端 / 调用方看的是 FastAPI 的 schema，
Agent 内部看的是 ADK 自己的 state / schema，
中间用 Runner 做桥接。

6. 如何跑起来 & 测试？

6.1 启动服务

在项目根目录（包含 my_agent/ 那层）执行：

uvicorn my_agent.api:app --reload --port8001

或直接运行 api.py 文件，看到类似日志说明启动成功：

INFO: Uvicorn running on http://0.0.0.0:8001(Press CTRL+C to quit)

6.2 创建会话

第一次聊天前，先创建一个 Session：

curl-X POST"http://127.0.0.1:8000/session"\-H"Content-Type: application/json"\-d'{"user_id": "user_123"}'

返回示例：

{"user_id":"user_123","session_id":"c2a4a3d8-2a4e-4c2c-87a0-xxxxxx"}

前端要把这个 session_id 存起来（例如存在 localStorage / cookie / 内存）。

6.3 在会话里聊天

后续每一轮对话都用这个 user_id + session_id：

curl-X POST"http://127.0.0.1:8000/chat"\-H"Content-Type: application/json"\-d'{ "user_id": "user_123", "session_id": "c2a4a3d8-2a4e-4c2c-87a0-xxxxxx", "message": "现在东京几点？" }'

你会收到类似：

{"user_id":"user_123","session_id":"c2a4a3d8-2a4e-4c2c-87a0-xxxxxx","reply":"东京现在是 10:30 AM。"}

再发第二条：

{"user_id":"user_123","session_id":"c2a4a3d8-2a4e-4c2c-87a0-xxxxxx","message":"那和上海差多少？"}

ADK 会自动在同一条 Session 里拼接上下文，让 Agent 有“短期记忆”。

{"user_id":"user_123","session_id":"c2a4a3d8-2a4e-4c2c-87a0-xxxxxx","reply":"东京和上海的当前时间都是上午10:30，所以它们之间没有时差。"}

6.4 通过 Swagger UI 测试

可以访问 http://localhost:8001/docs 通过 UI 来进行会话测试。

7. 小结

我们这篇文章做了几件实打实的事：

• 在已有的 root_agent 基础上
• 新增 api.py，用 FastAPI + Uvicorn 暴露 HTTP API
• 用 InMemorySessionService + Runner 支持 user_id + session_id 的对话会话
• 拆解了 Session / Runner / FastAPI / GenAI Content 等关键技术点

一句话总结就是：

用 ADK 写 Agent，用 FastAPI 做壳，用 SessionService + Runner 管会话，
你就从“玩具 demo”迈进了“真正可被任何服务调用的 Agent 服务”。