AgentkitAgentServerApp是AgentKit SDK中基于A2A（Agent-to-Agent）协议的高级应用框架，用于构建功能完整的Agent服务端应用。它集成了Google ADK Web Server，提供完整的REST API端点，支持会话管理、流式对话、多模态输入、评估集等功能。

使用场景

需要完整ADK功能的Agent应用。

Agent间通信场景。

生产环境部署，需要完整的可观测性和生命周期管理能力。

多模态交互场景，需处理文本、图片、文件等多种输入类型。

核心特性

完整的ADK集成：提供会话管理、长期记忆、工件管理、凭证服务、评估集管理等功能。

A2A协议支持：将Agent封装为A2A服务，支持Agent之间的互相通信。

内置遥测功能：自动收集链路追踪、性能指标和日志信息。

生命周期管理：自动管控启动、运行、关闭全流程，完成资源初始化与清理工作。

架构优势

统一的API标准：遵循Google ADK的API规范。

多种调用方式：支持REST API、SSE流式。

完整的会话管理：自动处理会话创建、更新、删除。

生产就绪：内置遥测、错误处理。

跨平台兼容：与其他语言的ADK实现保持一致。

多模态支持：支持文本、图片、文件、函数调用等多种输入类型。

与其他应用框架对比

框架	用途	核心特性
AgentkitSimpleApp	标准Agent应用	简单易用，@app.entrypoint装饰器。
AgentkitMCPApp	MCP工具服务	@app.tool装饰器，工具封装。
AgentkitA2aApp	A2A通信	@app.agent_executor装饰器。
AgentkitAgentServerApp	完整服务端	集成ADK Web Server，支持A2A。

快速开始

创建并启动服务器。

import os
from agentkit.apps import AgentkitAgentServerApp
from veadk import Agent
from veadk.memory.short_term_memory import ShortTermMemory

from veadk.tracing.telemetry.exporters.apmplus_exporter import APMPlusExporter
from veadk.tracing.telemetry.opentelemetry_tracer import OpentelemetryTracer

exporters = [
APMPlusExporter( ) if os.getenv("ENABLE_APMPLUS", "").lower() == "true" else None,
]
tracer = OpentelemetryTracer(
exporters=[exporter for exporter in exporters if exporter is not None]
)

# 创建 Agent
root_agent = Agent(
name="root_agent",
instruction="您是一个乐于助人的助手。您可以回答用户的问题。",
tracers=[tracer]
)

# 创建短期记忆服务
short_term_memory = ShortTermMemory(backend="local")

# 创建应用
agent_server_app = AgentkitAgentServerApp(
agent=root_agent,
short_term_memory=short_term_memory,
)

# 启动服务器
if __name__ == "__main__":
agent_server_app.run(host="0.0.0.0", port=8000)

测试Agent。

创建会话。

curl -X POST http://localhost:8000/apps/root_agent/users/u_123/sessions/s_123 \
-H "Content-Type: application/json" \
-d '{}'

发送流式对话请求。

curl -X POST http://localhost:8000/run_sse \
-H "Content-Type: application/json" \
-d '{
"app_name": "root_agent",
"user_id": "u_123",
"session_id": "s_123",
"streaming": false,
"new_message": {
"role": "user",
"parts": [{"text": "介绍一下您自己"}]
}
}'

若返回以下格式的响应，说明Agent运行正常。

{
"data": {
"modelVersion": "doubao-seed-1-6-250615",
"content": {
"parts": [
{
"text": "用户让我再次介绍一下自己。首先，我需要回顾之前的对话历史，看看之前是怎么回答的，避免重复但也要保持一致性。之前的回答提到了我是VeADK团队开发的AI助手，专注于数据科学、文档撰写和软件开发，能提供问题解答、技术支持和信息咨询。现在用户再次提问，可能是想确认信息，或者希望得到更详细的介绍。我需要保持核心信息不变，同时可以稍微扩展一下，比如提到具体能帮助用户做什么，比如数据分析、代码编写、文档整理等，让介绍更具体。另外，要注意语气友好，符合乐于助人的助手定位。需要确保不添加新的、未在system prompt中提到的信息。system prompt里说我是VeADK团队开发的，专注于数据科学、文档、软件开发，所以这些是重点。可以强调在这些领域的专业性，以及如何帮助用户高效解决问题。避免使用复杂术语，保持简洁明了。总结一下，回答的结构应该是：团队背景、专注领域、具体帮助内容、目标（助力解决问题）。这样既回应了用户的请求，又保持了一致性和专业性。",
"thought": true
},
{
"text": "我是由VeADK团队开发的AI助手，核心定位是在数据科学、文档撰写与软件开发领域提供专业支持。在数据科学方向，我能协助处理数据分析、模型理解、可视化方案设计等问题；文档撰写上，可帮你梳理逻辑框架、优化表达、整理技术文档或报告；软件开发领域，支持代码解读、基础功能实现建议、开发流程咨询等。我的目标是通过精准的信息输出和实用的解决方案，帮你高效应对技术场景中的各类需求。"
}
],
"role": "model"
},
"partial": false,
"finishReason": "STOP",
"usageMetadata": {
"cachedContentTokenCount": 0,
"candidatesTokenCount": 363,
"promptTokenCount": 209,
"totalTokenCount": 572
},
"invocationId": "e-a78b1c29-908e-4cfd-8e0e-e4822ad0****",
"author": "root_agent",
"actions": {
"stateDelta": {},
"artifactDelta": {},
"requestedAuthConfigs": {},
"requestedToolConfirmations": {}
},
"id": "b4cef459-4ad7-42bd-aa9d-b3040391****",
"timestamp": 1768825786.93406
}
}

API端点详解

会话管理端点（/sessions）

用于创建Agent会话，支持两种创建方式：

方式一：指定session_id

创建指定session_id的Agent会话。

端点：POST /apps/{app_name}/users/{user_id}/sessions/{session_id}

请求示例:

curl -X POST http://localhost:8000/apps/root_agent/users/u_123/sessions/s_123 \
-H "Content-Type: application/json" \
-d '{}'

响应示例:

{
"id": "s_123",
"appName": "root_agent",
"userId": "u_123",
"state": {},
"events": [],
"lastUpdateTime": 1743711430.022186
}

方式二：自动分配session_id

调用后系统自动生成唯一session_id创建新的Agent会话，每次调用均会生成全新会话实例。适用于无需手动管理session_id的业务场景。

端点：POST /apps/{app_name}/users/{user_id}/sessions

请求示例：

curl -X POST http://localhost:8000/apps/root_agent/users/u_123/sessions \
-H "Content-Type: application/json" \
-d '{}'

对话请求端点（/run_sse）

用于发送对话请求，支持SSE流式响应或一次性响应，推荐使用/run_sse。

端点：POST /run_sse（推荐，流式）或POST /run（非流式）

请求字段说明

参数	类型	必填	说明
app_name	string	是	Agent应用名称。
user_id	string	是	用户唯一标识。
session_id	string	否	会话唯一标识（创建会话时可省略）。
new_message	Content	是	用户消息内容，支持多模态。具体请参见“new_message”结构。
streaming	boolean	否	是否使用流式响应，默认false。
state_delta	dict	否	状态更新（可选）。
invocation_id	string	否	调用ID，用于恢复长时间运行的函数（可选）。

new_message结构

参数	类型	必填	说明
parts	list[Part]	是	消息内容部分列表，每个part可以是不同类型（文本、图片、文件等）。具体请参见“Part”结构。
role	string	否	内容的生产者，必须是user或model，用于多轮对话。如果不设置，系统会自动推断。 user：用户。 model：模型。

Part结构

Python代码	JSON字段	类型	说明
types.Part(text="...")	{"text": "..."}	String	纯文本内容。
types.Part(inline_data=...)	{"inlineData": {...}}	Blob	内联数据（如Base64编码的图片、视频等）。
types.Part(file_data=...)	{"fileData": {...}}	FileData	文件引用（通过URI访问）。
types.Part(function_call=...)	{"functionCall": {...}}	FunctionCall	函数调用。
types.Part(function_response=...)	{"functionResponse": {...}}	FunctionResponse	函数响应。
types.Part(executable_code=...)	{"executableCode": {...}}	ExecutableCode	可执行代码。
types.Part(code_execution_result=...)	{"codeExecutionResult": {...}}	CodeExecutionResult	代码执行结果。
types.Part(thought=...)	{"thought": true/false}	Boolean	指示是否为模型的思考。
types.Part(thought_signature=...)	{"thoughtSignature": "..."}	Bytes	思考的签名，用于后续请求。
types.Part(media_resolution=...)	{"mediaResolution": {...}}	PartMediaResolution	输入媒体的分辨率。
types.Part(video_metadata=...)	{"videoMetadata": {...}}	VideoMetadata	视频元数据（仅在提供视频数据时使用）。

不同类型消息请求示例

纯文本消息

Python示例：

from google.genai import types

agent = Agent()
runner = Runner(agent=agent)

message = types.UserContent(
parts=[
types.Part(text="介绍一下您自己")
]
)

async for event in runner.run_async(
user_id=runner.user_id,session_id=session_id,new_message=message,
):
...

curl示例：

curl -X POST http://localhost:8000/run_sse \
-H "Content-Type: application/json" \
-d '{
"app_name": "root_agent",
"user_id": "u_123",
"session_id": "s_123",
"streaming": true,
"new_message": {
"role": "user",
"parts": [{"text": "介绍一下您自己"}]
}
}'

多模态消息：文本和Base64编码图片

Python示例：

from google.genai import types
import base64

# 读取图片并转换为 base64
with open("image.png", "rb") as f:
image_data = base64.b64encode(f.read()).decode()

message = types.UserContent(
parts=[
types.Part(text="描述一下这张图片"),
types.Part(
inline_data=types.Blob(
data=base64.b64decode(image_data),
mime_type="image/png"
)
)
]
)

curl示例：

curl -X POST http://localhost:8000/run \
-H "Content-Type: application/json" \
-d '{
"app_name": "root_agent",
"user_id": "u_123",
"session_id": "s_123",
"streaming": false,
"new_message": {
"role": "user",
"parts": [
{
"text": "描述一下这张图片"
},
{
"inlineData": {
"data": "iVBORw0KGgoAAAANSUhEUgAAAgAAIACAYAAAD0eNT6AAAACXBIWXMAAAsTAAALEwEAmpw...",
"mimeType": "image/png"
}
}
]
}
}'

多模态消息：文本和图片

Python示例：

from google.genai import types

message = types.UserContent(
parts=[
types.Part(text="描述一下这张图片"),
types.Part(
file_data=types.FileData(
display_name="image1.png",
file_uri="file:///path/to/image.png",
mime_type="image/png"
)
)
]
)

curl示例：

curl -X POST http://localhost:8000/run \
-H "Content-Type: application/json" \
-d '{
"app_name": "root_agent",
"user_id": "u_123",
"session_id": "s_123",
"streaming": false,
"new_message": {
"role": "user",
"parts": [
{
"text": "描述一下这张图片"
},
{
"fileData": {
"displayName": "image1.png",
"fileUri": "file:///path/to/image.png",
"mimeType": "image/png"
}
}
]
}
}'

多模态消息：函数调用

Python示例：

from google.genai import types

message = types.UserContent(
parts=[
types.Part(
function_call=types.FunctionCall(
name="get_weather",
args={"city": "北京"}
)
)
]
)

curl示例：

curl -X POST http://localhost:8000/run \
-H "Content-Type: application/json" \
-d '{
"app_name": "root_agent",
"user_id": "u_123",
"session_id": "s_123",
"streaming": false,
"new_message": {
"role": "user",
"parts": [
{
"functionCall": {
"name": "get_weather",
"args": {"city": "北京"}
}
}
]
}
}'

响应格式

流式响应（streaming: true）：

data: {"author": "agent", "parts": [{"text": "好的，"}]}

data: {"author": "agent", "parts": [{"text": "我来帮您"}]}

data: {"author": "agent", "parts": [{"text": "描述这张图片"}]}

data: {"author": "agent", "parts": [{"text": "..."}]}

data: {"author": "agent", "parts": [{"text": "完成！"}]}

非流式响应（streaming: false）：

{"author": "agent", "parts": [{"text": "好的，我来帮您描述这张图片。"}]}, # 一次性返回

兼容性端点（/invoke）

简化版对话端点，兼容AgentKit CLI调用方式。

端点：POST /invoke。

特点:

从HTTP headers中提取user_id和session_id。

支持prompt字段（简化版，自动转换为new_message）。

自动创建会话（如果不存在）。

始终返回SSE流式响应。

请求示例：

curl -X POST http://localhost:8000/invoke \
-H "user_id: u_123" \
-H "session_id: s_123" \
-H "Content-Type: application/json" \
-d '{
"prompt": "介绍一下您自己"
}'

参数说明：

Headers：

user_id或x-user-id：用户唯一标识。

session_id：会话唯一标识。

Body：

prompt：文本提示（自动转换为new_message）。

响应格式：

data: {"author": "agent", "parts": [{"text": "您好！"}]}

data: {"author": "agent", "parts": [{"text": "我是一个..."}]}

data: {"author": "agent", "parts": [{"text": "..."}]}