92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
"""
|
||
Dev Agent A2A Server
|
||
使用 ADK 官方 A2A 协议暴露 Dev Agent，支持 HTTP 调用，可直接上云部署。
|
||
|
||
启动方式：
|
||
python a2a_server.py
|
||
|
||
端点：
|
||
A2A 接口: http://127.0.0.1:8001/a2a/dev_agent
|
||
Agent卡片: http://127.0.0.1:8001/.well-known/agent-card.json
|
||
健康检查: http://127.0.0.1:8001/health
|
||
"""
|
||
import os
|
||
import sys
|
||
|
||
# 确保项目根目录在 path 里
|
||
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
|
||
if PROJECT_ROOT not in sys.path:
|
||
sys.path.insert(0, PROJECT_ROOT)
|
||
|
||
from dotenv import load_dotenv
|
||
load_dotenv(os.path.join(PROJECT_ROOT, "agents/my_agent", ".env"))
|
||
|
||
# 强制 UTF-8
|
||
os.environ["PYTHONUTF8"] = "1"
|
||
|
||
import uvicorn
|
||
from google.adk.a2a.utils.agent_to_a2a import to_a2a
|
||
from google.adk.runners import Runner
|
||
from google.adk.sessions.sqlite_session_service import SqliteSessionService
|
||
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
|
||
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
|
||
from agents.my_agent.app import dev_app
|
||
|
||
|
||
# 配置
|
||
HOST = os.getenv("A2A_SERVER_HOST", "0.0.0.0") # 监听地址
|
||
PORT = int(os.getenv("A2A_SERVER_PORT", "8001"))
|
||
# Agent Card 中对外公布的地址（客户端用这个来连接，不能用 0.0.0.0）
|
||
PUBLIC_HOST = os.getenv("A2A_PUBLIC_HOST", "127.0.0.1")
|
||
|
||
# 数据目录
|
||
DATA_DIR = os.path.join(PROJECT_ROOT, "data")
|
||
os.makedirs(DATA_DIR, exist_ok=True)
|
||
|
||
# --- Session 服务：SQLite 持久化（重启不丢失对话历史）---
|
||
session_service = SqliteSessionService(
|
||
db_path=os.path.join(DATA_DIR, "sessions.db")
|
||
)
|
||
|
||
# --- Memory 服务：长期记忆（先用内存版，后续可换 Chroma 等向量库）---
|
||
memory_service = InMemoryMemoryService()
|
||
|
||
# --- Artifact 服务：工件存储（大文件等）---
|
||
artifact_service = InMemoryArtifactService()
|
||
|
||
# --- 构建 Runner：会话 + 记忆 + 压缩 一体化 ---
|
||
runner = Runner(
|
||
app=dev_app,
|
||
session_service=session_service,
|
||
memory_service=memory_service,
|
||
artifact_service=artifact_service,
|
||
auto_create_session=True,
|
||
)
|
||
|
||
# 用 ADK 官方工具把 agent 转成 A2A 服务
|
||
# 传入自定义 runner，启用 SQLite 持久化 + 上下文压缩 + 记忆
|
||
a2a_app = to_a2a(
|
||
agent=dev_app.root_agent,
|
||
host=PUBLIC_HOST, # agent card 里用的对外地址
|
||
port=PORT,
|
||
runner=runner,
|
||
)
|
||
|
||
|
||
def main():
|
||
print("=" * 60)
|
||
print("Dev Agent A2A Server 启动中...")
|
||
print(f" 监听地址: http://{HOST}:{PORT}")
|
||
print(f" 对外地址: http://{PUBLIC_HOST}:{PORT}")
|
||
print(f" A2A 端点: http://{PUBLIC_HOST}:{PORT}/")
|
||
print(f" Agent 卡片: http://{PUBLIC_HOST}:{PORT}/.well-known/agent-card.json")
|
||
print(f" 会话持久化: SQLite ({DATA_DIR}/sessions.db)")
|
||
print(f" 上下文压缩: 每 20 轮自动摘要")
|
||
print("=" * 60)
|
||
|
||
uvicorn.run(a2a_app, host=HOST, port=PORT, log_level="info")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|