synapse/docs/02-guide/grpc-transport.md

灵隙协议实现规范：gRPC 传输协议

版本：1.0.0 | 层级：实现层

---

1. Proto 定义位置说明

Synapse 协议的 gRPC 服务定义统一存放于以下位置：

proto/
├── synapse.proto              # 主 proto 文件，包含所有服务与消息定义
├── synapse_common.proto       # 公共消息类型（可选拆分）
├── google/
│   └── api/
│       ├── annotations.proto  # Google HTTP 注解依赖
│       └── http.proto
└── protoc-gen-openapiv2/
    └── options/
        └── annotations.proto  # OpenAPI v2 注解依赖（可选）

主定义文件 synapse.proto 包含：

• 5 个 gRPC 服务（TaskService、SynapseMessageService、CapabilitiesService、GroupService、HealthService）
• 20 个 RPC 方法
• 对应的请求/响应消息类型
• google.api.http 注解，用于通过 gRPC Gateway 生成 HTTP REST 端点

通过 buf 或 protoc 工具链编译生成：

生成产物	说明
synapse.pb.go	gRPC 服务端/客户端桩代码
synapse.pb.gw.go	gRPC Gateway 反向代理代码
synapse.swagger.json	OpenAPI v2 规范文档（可选）

---

2. 服务定义

2.1 TaskService（任务服务）

任务服务是协议核心，负责任务的提交、查询、取消与状态流式推送。

service TaskService {
  // 提交任务（JSON payload）
  rpc SubmitTask(SubmitTaskRequest) returns (SubmitTaskResponse);

  // 提交任务（含文件附件，Multipart）
  rpc SubmitTaskWithFiles(SubmitTaskWithFilesRequest) returns (SubmitTaskResponse);

  // 获取任务完整信息
  rpc GetTask(GetTaskRequest) returns (TaskStatusResponse);

  // 获取任务状态（轻量查询）
  rpc GetTaskStatus(GetTaskRequest) returns (TaskStatusResponse);

  // 获取任务列表（支持分页和过滤）
  rpc ListTasks(ListTasksRequest) returns (ListTasksResponse);

  // 取消任务
  rpc CancelTask(CancelTaskRequest) returns (CancelTaskResponse);

  // 流式监听任务状态变化（双向流）
  rpc StreamTaskStatus(stream TaskStatusStreamRequest) returns (stream TaskStatusUpdate);
}

RPC 方法列表：

#	方法名	请求类型	响应类型	流式
1	SubmitTask	SubmitTaskRequest	SubmitTaskResponse	否
2	SubmitTaskWithFiles	SubmitTaskWithFilesRequest	SubmitTaskResponse	否
3	GetTask	GetTaskRequest	TaskStatusResponse	否
4	GetTaskStatus	GetTaskRequest	TaskStatusResponse	否
5	ListTasks	ListTasksRequest	ListTasksResponse	否
6	CancelTask	CancelTaskRequest	CancelTaskResponse	否
7	StreamTaskStatus	stream TaskStatusStreamRequest	stream TaskStatusUpdate	双向流

2.2 SynapseMessageService（消息服务）

消息服务负责 Agent 间的异步消息传递、流式对话以及回调推送。

service SynapseMessageService {
  // 发送单条消息
  rpc SendMessage(SendMessageRequest) returns (SendMessageResponse);

  // 流式消息交换（双向流，适用于多轮对话）
  rpc StreamMessages(stream SynapseMessage) returns (stream SynapseMessage);

  // 接收服务端推送的回调消息（服务器推送流）
  rpc ReceiveCallback(google.protobuf.Empty) returns (stream SynapseMessage);
}

RPC 方法列表：

#	方法名	请求类型	响应类型	流式
8	SendMessage	SendMessageRequest	SendMessageResponse	否
9	StreamMessages	stream SynapseMessage	stream SynapseMessage	双向流
10	ReceiveCallback	google.protobuf.Empty	stream SynapseMessage	服务端流

2.3 CapabilitiesService（能力服务）

能力服务提供 Agent 能力查询与发现机制。

service CapabilitiesService {
  // 获取当前 Agent 自身的能力声明
  rpc GetCapabilities(google.protobuf.Empty) returns (CapabilitiesResponse);

  // 查询指定 Agent 的能力
  rpc QueryAgent(QueryAgentRequest) returns (CapabilitiesResponse);

  // 列出网络中所有活跃 Agent
  rpc ListAgents(google.protobuf.Empty) returns (ListAgentsResponse);
}

RPC 方法列表：

#	方法名	请求类型	响应类型	流式
11	GetCapabilities	google.protobuf.Empty	CapabilitiesResponse	否
12	QueryAgent	QueryAgentRequest	CapabilitiesResponse	否
13	ListAgents	google.protobuf.Empty	ListAgentsResponse	否

2.4 GroupService（群组服务）

群组服务支持 Agent 群的创建、成员管理与群内消息流。

service GroupService {
  // 创建群组
  rpc CreateGroup(CreateGroupRequest) returns (CreateGroupResponse);

  // 加入群组
  rpc JoinGroup(JoinGroupRequest) returns (JoinGroupResponse);

  // 离开群组
  rpc LeaveGroup(LeaveGroupRequest) returns (LeaveGroupResponse);

  // 解散群组（仅群主/管理员）
  rpc DisbandGroup(DisbandGroupRequest) returns (DisbandGroupResponse);

  // 列出群组成员
  rpc ListGroupMembers(ListGroupMembersRequest) returns (ListGroupMembersResponse);

  // 流式收发群组消息（双向流）
  rpc StreamGroupMessages(stream SynapseMessage) returns (stream SynapseMessage);
}

RPC 方法列表：

#	方法名	请求类型	响应类型	流式
14	CreateGroup	CreateGroupRequest	CreateGroupResponse	否
15	JoinGroup	JoinGroupRequest	JoinGroupResponse	否
16	LeaveGroup	LeaveGroupRequest	LeaveGroupResponse	否
17	DisbandGroup	DisbandGroupRequest	DisbandGroupResponse	否
18	ListGroupMembers	ListGroupMembersRequest	ListGroupMembersResponse	否
19	StreamGroupMessages	stream SynapseMessage	stream SynapseMessage	双向流

2.5 HealthService（健康服务）

健康服务提供存活探针与健康状态检查。

service HealthService {
  // 健康检查
  rpc HealthCheck(google.protobuf.Empty) returns (HealthCheckResponse);
}

RPC 方法列表：

#	方法名	请求类型	响应类型	流式
20	HealthCheck	google.protobuf.Empty	HealthCheckResponse	否

---

3. HTTP 路由映射表

通过 google.api.http 注解，每个 gRPC 方法映射为对应的 HTTP REST 端点。gRPC Gateway 在启动时读取这些注解，自动生成反向代理，使同一套后端可同时服务于 gRPC 和 HTTP 客户端。

3.1 完整映射表

#	gRPC 方法	HTTP 方法	HTTP 路径	说明
1	SubmitTask	POST	/api/v1/synapse/task	提交任务
2	SubmitTaskWithFiles	POST	/api/v1/synapse/task/multipart	提交含附件的任务
3	GetTask	GET	/api/v1/synapse/task/{task_id}	获取任务详情
4	GetTaskStatus	GET	/api/v1/synapse/task/{task_id}/status	获取任务状态
5	ListTasks	GET	/api/v1/synapse/tasks	列出任务（支持分页）
6	CancelTask	POST	/api/v1/synapse/task/{task_id}/cancel	取消任务
7	StreamTaskStatus	GET	/api/v1/synapse/task/{task_id}/stream	SSE 流式监听任务状态
8	SendMessage	POST	/api/v1/synapse/message	发送消息
9	StreamMessages	GET	/api/v1/synapse/messages/stream	WebSocket/SSE 流式消息交换
10	ReceiveCallback	GET	/api/v1/synapse/messages/callback	SSE 接收服务端回调推送
11	GetCapabilities	GET	/api/v1/synapse/capabilities	获取自身能力声明
12	QueryAgent	GET	/api/v1/synapse/agents/{agent_id}	查询指定 Agent 能力
13	ListAgents	GET	/api/v1/synapse/agents	列出所有活跃 Agent
14	CreateGroup	POST	/api/v1/synapse/group	创建群组
15	JoinGroup	POST	/api/v1/synapse/group/{group_id}/join	加入群组
16	LeaveGroup	POST	/api/v1/synapse/group/{group_id}/leave	离开群组
17	DisbandGroup	POST	/api/v1/synapse/group/{group_id}/disband	解散群组
18	ListGroupMembers	GET	/api/v1/synapse/group/{group_id}/members	列出群组成员
19	StreamGroupMessages	GET	/api/v1/synapse/group/{group_id}/messages/stream	WebSocket/SSE 流式群组消息
20	HealthCheck	GET	/api/v1/synapse/health	健康检查

3.2 路由特性说明

路径参数绑定：

路径参数	绑定字段	示例值
{task_id}	GetTaskRequest.task_id	"task-abc123"
{agent_id}	QueryAgentRequest.agent_id	"agent-xyz789"
{group_id}	JoinGroupRequest.group_id	"group-demo-001"

请求体映射规则：

• POST 请求：除路径参数外，剩余字段映射为 HTTP 请求体（JSON）
• GET 请求：非路径参数字段映射为 URL 查询参数（query string）

流式端点说明：

流式 RPC	HTTP 协议映射	说明
StreamTaskStatus	SSE（Server-Sent Events）	服务端单向推送任务状态变更
StreamMessages	WebSocket	全双工流式消息交换
ReceiveCallback	SSE（Server-Sent Events）	服务端单向推送回调消息
StreamGroupMessages	WebSocket	全双工群组消息流

对于双向流 RPC（StreamMessages、StreamGroupMessages），HTTP REST 映射通过 WebSocket Upgrade 实现；对于服务端流 RPC（StreamTaskStatus、ReceiveCallback），通过 SSE（text/event-stream）实现。

3.3 路由前缀约定

所有 Synapse 协议 REST API 遵循以下 URL 前缀约定：

/api/v1/synapse/

• api：标识为 API 请求
• v1：API 版本号
• synapse：命名空间，区分 Synapse 协议与其他业务 API

版本升级时（如 v1 → v2），路径前缀同步更新为 /api/v2/synapse/，新旧版本可并行部署。

3.4 HTTP 状态码与 gRPC 状态码映射

HTTP 状态码	gRPC 状态码	说明
200 OK	OK	请求成功
201 Created	OK	资源创建成功（SubmitTask、CreateGroup）
400 Bad Request	InvalidArgument	请求参数无效
401 Unauthorized	Unauthenticated	认证失败
403 Forbidden	PermissionDenied	权限不足
404 Not Found	NotFound	资源不存在
409 Conflict	AlreadyExists / Aborted	资源冲突
429 Too Many Requests	ResourceExhausted	限流触发
500 Internal Server Error	Internal	服务端内部错误
503 Service Unavailable	Unavailable	服务不可用

gRPC Gateway 自动完成上述状态码转换，客户端无需额外处理。

---

4. 协议缓冲区消息结构概览

以下列出各服务核心 RPC 的请求与响应消息结构（仅示意字段分类，完整定义以 .proto 文件为准）：

4.1 任务相关消息

// SubmitTask 请求
message SubmitTaskRequest {
  string task_type = 1;           // 任务类型
  string payload_json = 2;        // JSON 负载
  string callback_url = 3;        // 回调地址（可选）
  int32  ttl_seconds = 4;         // 任务 TTL（可选）
  string idempotency_key = 5;     // 幂等键（可选）
}

// SubmitTask 响应
message SubmitTaskResponse {
  string task_id = 1;             // 任务 ID
  string status = 2;              // 初始状态
  string message = 3;             // 附加信息
}

// 任务状态查询响应（通用）
message TaskStatusResponse {
  string task_id = 1;
  string status = 2;
  int32  progress = 3;            // 0-100
  string result_json = 4;         // 执行结果（JSON）
  string error_message = 5;
  string created_at = 6;
  string updated_at = 7;
  string etag = 8;                // 用于增量查询
}

// 任务流式更新
message TaskStatusUpdate {
  string task_id = 1;
  string event_type = 2;          // status_update / progress / log / error / complete
  string data_json = 3;
}

4.2 消息相关消息

// SendMessage 请求
message SendMessageRequest {
  string receiver_id = 1;
  string content_type = 2;        // text / json / code / file
  string payload_json = 3;
  string correlation_id = 4;      // 关联 ID（可选）
  string conversation_id = 5;     // 会话 ID（可选）
}

// SendMessage 响应
message SendMessageResponse {
  string message_id = 1;
  string status = 2;
}

// 统一消息信封（用于流式传输）
message SynapseMessage {
  string id = 1;
  string sender_id = 2;
  string receiver_id = 3;
  string content_type = 4;
  string payload_json = 5;
  string correlation_id = 6;
  string conversation_id = 7;
  int64  timestamp = 8;
}

4.3 能力相关消息

message CapabilitiesResponse {
  string agent_id = 1;
  string agent_name = 2;
  string protocol_version = 3;
  repeated string supported_task_types = 4;
  repeated string supported_content_types = 5;
  int64  max_file_size = 6;
  int32  max_concurrent_tasks = 7;
  bool   support_streaming = 8;
  bool   support_hil = 9;
  bool   support_groups = 10;
}

message QueryAgentRequest {
  string agent_id = 1;
}

message ListAgentsResponse {
  repeated AgentInfo agents = 1;
  int32 total = 2;
}

message AgentInfo {
  string agent_id = 1;
  string agent_name = 2;
  string endpoint = 3;
  repeated string capabilities = 4;
  string status = 5;              // online / offline / busy
}

4.4 群组相关消息

message CreateGroupRequest {
  string name = 1;
  string description = 2;
  repeated string initial_members = 3;
}

message CreateGroupResponse {
  string group_id = 1;
  string status = 2;
}

message JoinGroupRequest {
  string group_id = 1;
  string agent_id = 2;
}

message LeaveGroupRequest {
  string group_id = 1;
  string agent_id = 2;
}

message DisbandGroupRequest {
  string group_id = 1;
  string agent_id = 2;            // 需群主或管理员权限
}

message ListGroupMembersRequest {
  string group_id = 1;
}

message ListGroupMembersResponse {
  string group_id = 1;
  repeated GroupMember members = 2;
}

message GroupMember {
  string agent_id = 1;
  string role = 2;                // owner / admin / member
  string joined_at = 3;
}

4.5 健康检查消息

message HealthCheckResponse {
  string status = 1;              // SERVING / NOT_SERVING
  string version = 2;
  int64  uptime_seconds = 3;
}

---

5. 传输配置与使用指引

5.1 gRPC 客户端连接配置

// 推荐 gRPC 客户端配置
conn, err := grpc.Dial(
    target,
    grpc.WithInsecure(),                    // 开发环境；生产环境使用 TLS
    grpc.WithDefaultCallOptions(
        grpc.MaxCallRecvMsgSize(10*1024*1024), // 最大接收消息 10MB
        grpc.MaxCallSendMsgSize(10*1024*1024), // 最大发送消息 10MB
    ),
    grpc.WithKeepaliveParams(keepalive.ClientParameters{
        Time:                30 * time.Second, // 每 30s 发送心跳
        Timeout:             10 * time.Second, // 心跳超时 10s
        PermitWithoutStream: true,             // 无活跃流时也发送心跳
    }),
)

5.2 HTTP REST 客户端调用示例

# SubmitTask
curl -X POST http://localhost:8080/api/v1/synapse/task \
  -H "Content-Type: application/json" \
  -d '{"task_type":"text_generation","payload_json":"...","callback_url":"..."}'

# GetTask
curl http://localhost:8080/api/v1/synapse/task/task-abc123

# 流式监听任务状态（SSE）
curl -N http://localhost:8080/api/v1/synapse/task/task-abc123/stream

# HealthCheck
curl http://localhost:8080/api/v1/synapse/health

5.3 幂等性支持

对于 SubmitTask 和 CreateGroup 等写操作，建议通过 idempotency_key 字段实现幂等：

• 客户端为每个写操作生成唯一幂等键（如 UUID v4）
• 服务端缓存幂等键及其响应，相同键的重复请求返回缓存结果
• 幂等键缓存 TTL 默认为 24 小时

---

附录：gRPC Gateway 集成方式

┌─────────────────────────────────────────────────────┐
│                   外部客户端                          │
│   gRPC Client              HTTP Client (REST)       │
└────────┬───────────────────────────────┬────────────┘
         │ gRPC (h2)                    │ HTTP/1.1
         ▼                              ▼
┌─────────────────┐          ┌──────────────────────┐
│  gRPC Server    │          │  gRPC Gateway        │
│  (直接服务)      │◄────────►│  (反向代理 / 协议转换) │
└────────┬────────┘          └──────────────────────┘
         │ 内部 gRPC 调用
         ▼
┌─────────────────────────────────────────────────────┐
│              Synapse 业务逻辑层                       │
│   TaskService / MessageService / Capabilities ...   │
└─────────────────────────────────────────────────────┘

推荐集成方式：

1. 在 Go 服务端使用 grpc-gateway 库，通过一行代码注册反向代理
2. 启动时同时监听两个端口（例：gRPC 端口 9090，HTTP 端口 8080）
3. HTTP 端口同时提供 REST API 和 Swagger UI（开发环境）

---

文档结束