diff --git a/cmd/picoclaw/internal/gateway/helpers.go b/cmd/picoclaw/internal/gateway/helpers.go index 6852ec7af..ad55fb1fb 100644 --- a/cmd/picoclaw/internal/gateway/helpers.go +++ b/cmd/picoclaw/internal/gateway/helpers.go @@ -13,6 +13,7 @@ import ( "github.com/sipeed/picoclaw/pkg/agent" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/channels" + _ "github.com/sipeed/picoclaw/pkg/channels/api" _ "github.com/sipeed/picoclaw/pkg/channels/dingtalk" _ "github.com/sipeed/picoclaw/pkg/channels/discord" _ "github.com/sipeed/picoclaw/pkg/channels/feishu" diff --git a/config/api.example.json b/config/api.example.json new file mode 100644 index 000000000..a868d34dc --- /dev/null +++ b/config/api.example.json @@ -0,0 +1,21 @@ +{ + "channels": { + "api": { + "enabled": true, + "token": "your-secret-token-here", + "allow_origins": ["*"], + "ping_interval": 30, + "read_timeout": 60, + "max_connections": 100, + "placeholder": { + "enabled": true, + "text": "Thinking... 💭" + }, + "streaming": true + } + }, + "gateway": { + "host": "0.0.0.0", + "port": 8080 + } +} \ No newline at end of file diff --git a/config/config.flutter.json b/config/config.flutter.json new file mode 100644 index 000000000..cc87d845f --- /dev/null +++ b/config/config.flutter.json @@ -0,0 +1,43 @@ +{ + "agents": { + "defaults": { + "workspace": "~/.picoclaw/workspace", + "restrict_to_workspace": true, + "model_name": "gpt4", + "max_tokens": 8192, + "temperature": 0.7, + "max_tool_iterations": 20, + "summarize_message_threshold": 20, + "summarize_token_percent": 75 + } + }, + "model_list": [ + { + "model_name": "gpt4", + "model": "openai/gpt-4o", + "api_key": "sk-your-openai-key", + "api_base": "https://api.openai.com/v1" + } + ], + "channels": { + "imessage": { + "enabled": false + }, + "api": { + "enabled": true, + "token": "flutter-test-token-123" + } + }, + "tools": { + "exec": { + "enabled": true + }, + "skills": { + "enabled": true + } + }, + "gateway": { + "host": "127.0.0.1", + "port": 18792 + } +} \ No newline at end of file diff --git a/pkg/channels/api/api.go b/pkg/channels/api/api.go new file mode 100644 index 000000000..8378df7ec --- /dev/null +++ b/pkg/channels/api/api.go @@ -0,0 +1,503 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/google/uuid" + "github.com/gorilla/websocket" + + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/identity" + "github.com/sipeed/picoclaw/pkg/logger" +) + +// apiConn represents a single WebSocket connection. +type apiConn struct { + id string + conn *websocket.Conn + sessionID string + writeMu sync.Mutex + closed atomic.Bool +} + +// writeJSON sends a JSON message to the connection with write locking. +func (ac *apiConn) writeJSON(v any) error { + if ac.closed.Load() { + return fmt.Errorf("connection closed") + } + ac.writeMu.Lock() + defer ac.writeMu.Unlock() + return ac.conn.WriteJSON(v) +} + +// close closes the connection. +func (ac *apiConn) close() { + if ac.closed.CompareAndSwap(false, true) { + ac.conn.Close() + } +} + +// APIChannel implements a WebSocket API channel for Flutter app integration. +// It supports streaming responses for real-time typewriter effect. +type APIChannel struct { + *channels.BaseChannel + config config.APIConfig + upgrader websocket.Upgrader + connections sync.Map // connID → *apiConn + connCount atomic.Int32 + ctx context.Context + cancel context.CancelFunc +} + +// NewAPIChannel creates a new API channel. +func NewAPIChannel(cfg config.APIConfig, messageBus *bus.MessageBus) (*APIChannel, error) { + if cfg.Token == "" { + return nil, fmt.Errorf("api token is required") + } + + base := channels.NewBaseChannel("api", cfg, messageBus, cfg.AllowFrom) + + allowOrigins := cfg.AllowOrigins + checkOrigin := func(r *http.Request) bool { + if len(allowOrigins) == 0 { + return true // allow all if not configured + } + origin := r.Header.Get("Origin") + for _, allowed := range allowOrigins { + if allowed == "*" || allowed == origin { + return true + } + } + return false + } + + return &APIChannel{ + BaseChannel: base, + config: cfg, + upgrader: websocket.Upgrader{ + CheckOrigin: checkOrigin, + ReadBufferSize: 1024, + WriteBufferSize: 4096, // Larger buffer for streaming + }, + }, nil +} + +// Start implements Channel. +func (c *APIChannel) Start(ctx context.Context) error { + logger.InfoC("api", "Starting API channel") + c.ctx, c.cancel = context.WithCancel(ctx) + c.SetRunning(true) + logger.InfoC("api", "API channel started") + return nil +} + +// Stop implements Channel. +func (c *APIChannel) Stop(ctx context.Context) error { + logger.InfoC("api", "Stopping API channel") + c.SetRunning(false) + + // Close all connections + c.connections.Range(func(key, value any) bool { + if ac, ok := value.(*apiConn); ok { + ac.close() + } + c.connections.Delete(key) + return true + }) + + if c.cancel != nil { + c.cancel() + } + + logger.InfoC("api", "API channel stopped") + return nil +} + +// WebhookPath implements channels.WebhookHandler. +func (c *APIChannel) WebhookPath() string { return "/api/" } + +// ServeHTTP implements http.Handler for the shared HTTP server. +func (c *APIChannel) ServeHTTP(w http.ResponseWriter, r *http.Request) { + path := strings.TrimPrefix(r.URL.Path, "/api") + + switch { + case path == "/ws" || path == "/ws/": + c.handleWebSocket(w, r) + case path == "/health" || path == "/health/": + c.handleHealth(w, r) + default: + http.NotFound(w, r) + } +} + +// handleHealth returns the health status of the API channel. +func (c *APIChannel) handleHealth(w http.ResponseWriter, r *http.Request) { + status := map[string]any{ + "status": "ok", + "running": c.IsRunning(), + "connections": c.connCount.Load(), + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(status) +} + +// Send implements Channel — sends a message to the appropriate WebSocket connection. +func (c *APIChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { + if !c.IsRunning() { + return channels.ErrNotRunning + } + + outMsg := newMessage(TypeMessageCreate, map[string]any{ + "content": msg.Content, + }) + + return c.broadcastToSession(msg.ChatID, outMsg) +} + +// EditMessage implements channels.MessageEditor. +func (c *APIChannel) EditMessage(ctx context.Context, chatID string, messageID string, content string) error { + outMsg := newMessage(TypeMessageUpdate, map[string]any{ + "message_id": messageID, + "content": content, + }) + return c.broadcastToSession(chatID, outMsg) +} + +// StartTyping implements channels.TypingCapable. +func (c *APIChannel) StartTyping(ctx context.Context, chatID string) (func(), error) { + startMsg := newMessage(TypeTypingStart, nil) + if err := c.broadcastToSession(chatID, startMsg); err != nil { + return func() {}, err + } + return func() { + stopMsg := newMessage(TypeTypingStop, nil) + c.broadcastToSession(chatID, stopMsg) + }, nil +} + +// SendPlaceholder implements channels.PlaceholderCapable. +func (c *APIChannel) SendPlaceholder(ctx context.Context, chatID string) (string, error) { + if !c.config.Placeholder.Enabled { + return "", nil + } + + text := c.config.Placeholder.Text + if text == "" { + text = "Thinking... 💭" + } + + msgID := uuid.New().String() + outMsg := newMessage(TypeMessageCreate, map[string]any{ + "content": text, + "message_id": msgID, + }) + + if err := c.broadcastToSession(chatID, outMsg); err != nil { + return "", err + } + + return msgID, nil +} + +// StreamChunk sends a streaming chunk to the client. +func (c *APIChannel) StreamChunk(chatID, messageID, chunk string) error { + outMsg := newMessage(TypeMessageStream, map[string]any{ + "message_id": messageID, + "chunk": chunk, + }) + return c.broadcastToSession(chatID, outMsg) +} + +// StreamDone signals the end of streaming. +func (c *APIChannel) StreamDone(chatID, messageID string) error { + outMsg := newMessage(TypeMessageDone, map[string]any{ + "message_id": messageID, + }) + return c.broadcastToSession(chatID, outMsg) +} + +// broadcastToSession sends a message to all connections with a matching session. +func (c *APIChannel) broadcastToSession(chatID string, msg APIMessage) error { + // chatID format: "api:" + sessionID := strings.TrimPrefix(chatID, "api:") + msg.SessionID = sessionID + + var sent bool + c.connections.Range(func(key, value any) bool { + ac, ok := value.(*apiConn) + if !ok { + return true + } + if ac.sessionID == sessionID { + if err := ac.writeJSON(msg); err != nil { + logger.DebugCF("api", "Write to connection failed", map[string]any{ + "conn_id": ac.id, + "error": err.Error(), + }) + } else { + sent = true + } + } + return true + }) + + if !sent { + return fmt.Errorf("no active connections for session %s: %w", sessionID, channels.ErrSendFailed) + } + return nil +} + +// handleWebSocket upgrades the HTTP connection and manages the WebSocket lifecycle. +func (c *APIChannel) handleWebSocket(w http.ResponseWriter, r *http.Request) { + if !c.IsRunning() { + http.Error(w, "channel not running", http.StatusServiceUnavailable) + return + } + + // Authenticate + if !c.authenticate(r) { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + + // Check connection limit + maxConns := c.config.MaxConnections + if maxConns <= 0 { + maxConns = 100 + } + if int(c.connCount.Load()) >= maxConns { + http.Error(w, "too many connections", http.StatusServiceUnavailable) + return + } + + conn, err := c.upgrader.Upgrade(w, r, nil) + if err != nil { + logger.ErrorCF("api", "WebSocket upgrade failed", map[string]any{ + "error": err.Error(), + }) + return + } + + // Determine session ID from query param or generate one + sessionID := r.URL.Query().Get("session_id") + if sessionID == "" { + sessionID = uuid.New().String() + } + + ac := &apiConn{ + id: uuid.New().String(), + conn: conn, + sessionID: sessionID, + } + + c.connections.Store(ac.id, ac) + c.connCount.Add(1) + + logger.InfoCF("api", "WebSocket client connected", map[string]any{ + "conn_id": ac.id, + "session_id": sessionID, + }) + + // Send connected message to client + connectedMsg := newMessage(TypeConnected, map[string]any{ + "session_id": sessionID, + }) + if err := ac.writeJSON(connectedMsg); err != nil { + logger.ErrorCF("api", "Failed to send connected message", map[string]any{ + "error": err.Error(), + }) + ac.close() + c.connections.Delete(ac.id) + c.connCount.Add(-1) + return + } + + go c.readLoop(ac) +} + +// authenticate checks the Bearer token from the Authorization header. +func (c *APIChannel) authenticate(r *http.Request) bool { + token := c.config.Token + if token == "" { + return false + } + + // Check Authorization header + auth := r.Header.Get("Authorization") + if after, ok := strings.CutPrefix(auth, "Bearer "); ok { + if after == token { + return true + } + } + + // Check query parameter only when explicitly allowed + if c.config.AllowTokenQuery { + if r.URL.Query().Get("token") == token { + return true + } + } + + return false +} + +// readLoop reads messages from a WebSocket connection. +func (c *APIChannel) readLoop(ac *apiConn) { + defer func() { + ac.close() + c.connections.Delete(ac.id) + c.connCount.Add(-1) + logger.InfoCF("api", "WebSocket client disconnected", map[string]any{ + "conn_id": ac.id, + "session_id": ac.sessionID, + }) + }() + + readTimeout := time.Duration(c.config.ReadTimeout) * time.Second + if readTimeout <= 0 { + readTimeout = 60 * time.Second + } + + _ = ac.conn.SetReadDeadline(time.Now().Add(readTimeout)) + ac.conn.SetPongHandler(func(appData string) error { + _ = ac.conn.SetReadDeadline(time.Now().Add(readTimeout)) + return nil + }) + + // Start ping ticker + pingInterval := time.Duration(c.config.PingInterval) * time.Second + if pingInterval <= 0 { + pingInterval = 30 * time.Second + } + go c.pingLoop(ac, pingInterval) + + for { + select { + case <-c.ctx.Done(): + return + default: + } + + _, rawMsg, err := ac.conn.ReadMessage() + if err != nil { + if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) { + logger.DebugCF("api", "WebSocket read error", map[string]any{ + "conn_id": ac.id, + "error": err.Error(), + }) + } + return + } + + _ = ac.conn.SetReadDeadline(time.Now().Add(readTimeout)) + + var msg APIMessage + if err := json.Unmarshal(rawMsg, &msg); err != nil { + errMsg := newError("invalid_message", "failed to parse message") + ac.writeJSON(errMsg) + continue + } + + c.handleMessage(ac, msg) + } +} + +// pingLoop sends periodic ping frames to keep the connection alive. +func (c *APIChannel) pingLoop(ac *apiConn, interval time.Duration) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-c.ctx.Done(): + return + case <-ticker.C: + if ac.closed.Load() { + return + } + ac.writeMu.Lock() + err := ac.conn.WriteMessage(websocket.PingMessage, nil) + ac.writeMu.Unlock() + if err != nil { + return + } + } + } +} + +// handleMessage processes an inbound API Protocol message. +func (c *APIChannel) handleMessage(ac *apiConn, msg APIMessage) { + switch msg.Type { + case TypePing: + pong := newMessage(TypePong, nil) + pong.ID = msg.ID + ac.writeJSON(pong) + + case TypeMessageSend: + c.handleMessageSend(ac, msg) + + default: + errMsg := newError("unknown_type", fmt.Sprintf("unknown message type: %s", msg.Type)) + ac.writeJSON(errMsg) + } +} + +// handleMessageSend processes an inbound message.send from a client. +func (c *APIChannel) handleMessageSend(ac *apiConn, msg APIMessage) { + content, _ := msg.Payload["content"].(string) + if strings.TrimSpace(content) == "" { + errMsg := newError("empty_content", "message content is empty") + ac.writeJSON(errMsg) + return + } + + sessionID := msg.SessionID + if sessionID == "" { + sessionID = ac.sessionID + } + + chatID := "api:" + sessionID + senderID := "api-user" + + peer := bus.Peer{Kind: "direct", ID: "api:" + sessionID} + + metadata := map[string]string{ + "platform": "api", + "session_id": sessionID, + "conn_id": ac.id, + } + + logger.DebugCF("api", "Received message", map[string]any{ + "session_id": sessionID, + "preview": truncate(content, 50), + }) + + sender := bus.SenderInfo{ + Platform: "api", + PlatformID: senderID, + CanonicalID: identity.BuildCanonicalID("api", senderID), + } + + if !c.IsAllowedSender(sender) { + return + } + + c.HandleMessage(c.ctx, peer, msg.ID, senderID, chatID, content, nil, metadata, sender) +} + +// truncate truncates a string to maxLen runes. +func truncate(s string, maxLen int) string { + runes := []rune(s) + if len(runes) <= maxLen { + return s + } + return string(runes[:maxLen]) + "..." +} \ No newline at end of file diff --git a/pkg/channels/api/init.go b/pkg/channels/api/init.go new file mode 100644 index 000000000..593a9b73f --- /dev/null +++ b/pkg/channels/api/init.go @@ -0,0 +1,13 @@ +package api + +import ( + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/config" +) + +func init() { + channels.RegisterFactory("api", func(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) { + return NewAPIChannel(cfg.Channels.API, b) + }) +} \ No newline at end of file diff --git a/pkg/channels/api/protocol.go b/pkg/channels/api/protocol.go new file mode 100644 index 000000000..622b58a06 --- /dev/null +++ b/pkg/channels/api/protocol.go @@ -0,0 +1,47 @@ +package api + +import "time" + +// Protocol message types. +const ( + // Client -> Server + TypeMessageSend = "message.send" + TypePing = "ping" + + // Server -> Client + TypeConnected = "connected" + TypeMessageCreate = "message.create" + TypeMessageUpdate = "message.update" + TypeMessageStream = "message.stream" // Streaming chunk + TypeMessageDone = "message.done" // Streaming complete + TypeTypingStart = "typing.start" + TypeTypingStop = "typing.stop" + TypeError = "error" + TypePong = "pong" +) + +// APIMessage is the wire format for all API Protocol messages. +type APIMessage struct { + Type string `json:"type"` + ID string `json:"id,omitempty"` + SessionID string `json:"session_id,omitempty"` + Timestamp int64 `json:"timestamp,omitempty"` + Payload map[string]any `json:"payload,omitempty"` +} + +// newMessage creates an APIMessage with the given type and payload. +func newMessage(msgType string, payload map[string]any) APIMessage { + return APIMessage{ + Type: msgType, + Timestamp: time.Now().UnixMilli(), + Payload: payload, + } +} + +// newError creates an error APIMessage. +func newError(code, message string) APIMessage { + return newMessage(TypeError, map[string]any{ + "code": code, + "message": message, + }) +} \ No newline at end of file diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index db03c261c..d536aa7f2 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -277,6 +277,9 @@ func (m *Manager) initChannels() error { m.initChannel("pico", "Pico") } + if m.config.Channels.API.Enabled && m.config.Channels.API.Token != "" { + m.initChannel("api", "API") + } if m.config.Channels.Imessage.Enabled { m.initChannel("imessage", "Imessage") diff --git a/pkg/config/config.go b/pkg/config/config.go index 678f05dad..dd1113603 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -243,6 +243,7 @@ type ChannelsConfig struct { WeComApp WeComAppConfig `json:"wecom_app"` WeComAIBot WeComAIBotConfig `json:"wecom_aibot"` Pico PicoConfig `json:"pico"` + API APIConfig `json:"api"` Imessage ImessageConfig `json:"imessage"` IRC IRCConfig `json:"irc"` } @@ -444,6 +445,21 @@ type PicoConfig struct { Placeholder PlaceholderConfig `json:"placeholder,omitempty"` } +// APIConfig configures the API channel for Flutter app integration. +type APIConfig struct { + Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_API_ENABLED"` + Token string `json:"token" env:"PICOCLAW_CHANNELS_API_TOKEN"` + AllowTokenQuery bool `json:"allow_token_query,omitempty"` + AllowOrigins []string `json:"allow_origins,omitempty"` + PingInterval int `json:"ping_interval,omitempty"` + ReadTimeout int `json:"read_timeout,omitempty"` + WriteTimeout int `json:"write_timeout,omitempty"` + MaxConnections int `json:"max_connections,omitempty"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_API_ALLOW_FROM"` + Placeholder PlaceholderConfig `json:"placeholder,omitempty"` + Streaming bool `json:"streaming,omitempty"` // Enable streaming responses +} + type ImessageConfig struct { Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_IMESSAGE_ENABLED"` AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_IMESSAGE_ALLOW_FROM"`