stream

package
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 27, 2026 License: BSD-3-Clause Imports: 15 Imported by: 0

Documentation

Overview

Package stream 提供 WebSocket 流式消息适配器， 将 OrchestratorCallbacks 转换为前端 agent-ui 可消费的 JSON 消息。

Package stream provides generic WebSocket streaming types and a StreamManager for agent-go based applications.

Index

Constants

View Source 
const (
	TypeStarted          = "started"
	TypeMessageStart     = "message_start"
	TypeContentDelta     = "content_delta"
	TypeReasoningDelta   = "reasoning_delta"
	TypeMessageEnd       = "message_end"
	TypeToolCallStart    = "tool_call_start"
	TypeToolCallDelta    = "tool_call_delta"
	TypeToolCallEnd      = "tool_call_end"
	TypeToolExecuting    = "tool_executing"
	TypeToolResult       = "tool_result"
	TypeApprovalRequired = "approval_required"
	TypeQuestionnaire    = "questionnaire_request"
	TypeTokenUsage       = "token_usage"
	TypeTurnComplete     = "turn_complete"
	TypeError            = "error"
	TypeTerminalOutput   = "terminal_output"
)

Variables

This section is empty.

Functions

This section is empty.

Types

type Adapter 

type Adapter struct {
	// contains filtered or unexported fields
}

Adapter 将 OrchestratorCallbacks 转换为 15 种 WebSocket JSON 消息类型。 消息格式兼容 agent-ui 前端。

func NewAdapter 

func NewAdapter(sender MessageSender) *Adapter

NewAdapter 创建流式适配器。

func (*Adapter) CreateCallbacks 

func (a *Adapter) CreateCallbacks() *agent.OrchestratorCallbacks

CreateCallbacks 创建 OrchestratorCallbacks，将所有回调转换为 WebSocket 消息。

type MessageHandler 

type MessageHandler func(conn *websocket.Conn, req *WsRequest) bool

MessageHandler is a function that handles a custom WebSocket message. Returns true if the message was handled (prevents default dispatch).

type MessageSender 

type MessageSender interface {
	SendJSON(v any) error
}

MessageSender 发送 WebSocket 消息的接口（由调用方实现）。

type Orchestrator 

type Orchestrator interface {
	Orchestrate(
		ctx context.Context,
		sessionID, message, model, providerID, mode, thinking string,
		approvalMode ftool.ApprovalMode,
		termID string,
		callbacks *OrchestratorCallbacks,
		includeProjectDocs ...bool,
	)
	ResolveApproval(approvalID string, approved bool)
	ResumeAfterApproval(approvalID string, approved bool)
}

Orchestrator is the interface for running an agent orchestration.

type OrchestratorCallbacks 

type OrchestratorCallbacks = fwagent.OrchestratorCallbacks

OrchestratorCallbacks re-exports the agent package callbacks.

type StreamManager 

type StreamManager struct {
	// contains filtered or unexported fields
}

StreamManager manages active streaming sessions and dispatches orchestration callbacks to a message channel. Application-specific behavior (terminals, backups, etc.) should be added via wrapping or injection in the application layer.

func NewStreamManager 

func NewStreamManager(orchestrator Orchestrator) *StreamManager

NewStreamManager creates a new StreamManager.

func (*StreamManager) CancelStream 

func (sm *StreamManager) CancelStream(streamID string)

CancelStream cancels a streaming session by ID.

func (*StreamManager) ResolveApproval 

func (sm *StreamManager) ResolveApproval(approvalID string, approved bool)

ResolveApproval resolves an approval request.

func (*StreamManager) ResolveQuestionnaire 

func (sm *StreamManager) ResolveQuestionnaire(questionnaireID, answersJSON string)

ResolveQuestionnaire resolves a questionnaire request.

func (*StreamManager) StartStream 

func (sm *StreamManager) StartStream(
	sessionID, message, model, providerID, mode, thinking, approvalMode string,
	includeProjectDocs ...bool,
) (string, <-chan StreamMessage, error)

StartStream starts a new streaming session. Returns streamID, message channel, and error.

type StreamMessage 

type StreamMessage struct {
	Type            string `json:"type"`
	StreamID        string `json:"stream_id,omitempty"`
	SessionID       string `json:"session_id,omitempty"`
	MsgID           string `json:"msg_id,omitempty"`
	Seq             int64  `json:"seq,omitempty"`
	TurnID          string `json:"turn_id,omitempty"`
	CallID          string `json:"call_id,omitempty"`
	ToolMsgID       string `json:"tool_msg_id,omitempty"`
	Delta           string `json:"delta,omitempty"`
	Content         string `json:"content,omitempty"`
	Reasoning       string `json:"reasoning,omitempty"`
	ToolName        string `json:"tool_name,omitempty"`
	ToolArgs        string `json:"tool_args,omitempty"`
	FilePath        string `json:"file_path,omitempty"`
	IsStderr        bool   `json:"is_stderr,omitempty"`
	Approved        bool   `json:"approved,omitempty"`
	ApprovalID      string `json:"approval_id,omitempty"`
	QuestionnaireID string `json:"questionnaire_id,omitempty"`
	Text            string `json:"text,omitempty"`
	BackupPath      string `json:"backup_path,omitempty"`
	Error           string `json:"error,omitempty"`
	Code            string `json:"code,omitempty"`
	Mode            string `json:"mode,omitempty"`
	Thinking        string `json:"thinking,omitempty"`
	Model           string `json:"model,omitempty"`
	ProviderID      string `json:"provider_id,omitempty"`
	Message         string `json:"message,omitempty"`
	RiskLevel       string `json:"risk_level,omitempty"`
}

StreamMessage is a WebSocket message sent from backend to frontend. The 15 message types (started, message_start, content_delta, reasoning_delta, message_end, tool_call_start/delta/end, tool_executing, tool_result, approval_required, questionnaire_request, token_usage, turn_complete, error, terminal_output) are the protocol contract between agent-go and agent-ui.

type StreamSession 

type StreamSession struct {
	ID           string
	SessionID    string
	Cancel       context.CancelFunc
	CreatedAt    int64
	ApprovalMode ftool.ApprovalMode
}

StreamSession manages a single streaming session.

type WSServer 

type WSServer struct {
	// contains filtered or unexported fields
}

WSServer provides a WebSocket server for agent streaming communication. It handles /ws (main agent stream) and /files/{idx}/{path} (file serving) routes. Application-specific routes (e.g., /ws/terminal/) should be added by wrapping in the application layer.

func NewWSServer 

func NewWSServer(sm *StreamManager, fileDirs ...string) *WSServer

NewWSServer creates a new WebSocket server.

func (*WSServer) AddFileServerDir 

func (ws *WSServer) AddFileServerDir(dir string)

AddFileServerDir adds a directory for file serving.

func (*WSServer) BaseURL 

func (ws *WSServer) BaseURL() string

BaseURL returns the HTTP server base URL.

func (*WSServer) Broadcast 

func (ws *WSServer) Broadcast(data any)

Broadcast sends a JSON message to all connected clients. data can be any JSON-serializable value; it will be marshaled internally.

func (*WSServer) ClearFileServerDirs 

func (ws *WSServer) ClearFileServerDirs()

ClearFileServerDirs clears all file server directories.

func (*WSServer) Port 

func (ws *WSServer) Port() int

Port returns the server port number.

func (*WSServer) ServeFile 

func (ws *WSServer) ServeFile(w http.ResponseWriter, r *http.Request)

ServeFile handles file serving for /files/{idx}/{path...}.

func (*WSServer) ServeWS 

func (ws *WSServer) ServeWS(w http.ResponseWriter, r *http.Request)

ServeWS handles WebSocket upgrade and message loop.

func (*WSServer) SetOnMessage 

func (ws *WSServer) SetOnMessage(h MessageHandler)

SetOnMessage registers a custom message handler hook. Called before the default dispatch; if it returns true, the message is considered handled.

func (*WSServer) Start 

func (ws *WSServer) Start() error

Start starts the WebSocket server on a random localhost port. Application-specific routes can be registered on the mux after calling Start. Use NewServeMux() and register routes before Serve.

func (*WSServer) Stop 

func (ws *WSServer) Stop()

Stop stops the server.

type WsRequest 

type WsRequest struct {
	Type               string `json:"type"`
	SessionID          string `json:"session_id,omitempty"`
	StreamID           string `json:"stream_id,omitempty"`
	Message            string `json:"message,omitempty"`
	Model              string `json:"model,omitempty"`
	ProviderID         string `json:"provider_id,omitempty"`
	Mode               string `json:"mode,omitempty"`
	Thinking           string `json:"thinking,omitempty"`
	ApprovalMode       string `json:"approval_mode,omitempty"`
	IncludeProjectDocs bool   `json:"include_project_docs,omitempty"`
	ApprovalID         string `json:"approval_id,omitempty"`
	Approved           bool   `json:"approved,omitempty"`
	FilePath           string `json:"file_path,omitempty"`
	BackupPath         string `json:"backup_path,omitempty"`
	QuestionnaireID    string `json:"questionnaire_id,omitempty"`
	Text               string `json:"text,omitempty"`
	MessageID          string `json:"message_id,omitempty"`
}

WsRequest is a WebSocket client request.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.