made /chat asynchronous
This commit is contained in:
parent
60f3b7419f
commit
4baf3d260f
3 changed files with 248 additions and 30 deletions
|
|
@ -566,6 +566,7 @@ For detailed guides beyond this README:
|
||||||
| [SubTurn](docs/subturn.md) | Subagent coordination, concurrency control, lifecycle |
|
| [SubTurn](docs/subturn.md) | Subagent coordination, concurrency control, lifecycle |
|
||||||
| [Troubleshooting](docs/troubleshooting.md) | Common issues and solutions |
|
| [Troubleshooting](docs/troubleshooting.md) | Common issues and solutions |
|
||||||
| [Tools Configuration](docs/tools_configuration.md) | Per-tool enable/disable, exec policies, MCP, Skills |
|
| [Tools Configuration](docs/tools_configuration.md) | Per-tool enable/disable, exec policies, MCP, Skills |
|
||||||
|
| [Gateway API Reference](docs/api.md) | HTTP endpoints: `/chat`, `/health`, `/ready`, `/reload` |
|
||||||
| [Hardware Compatibility](docs/hardware-compatibility.md) | Tested boards, minimum requirements |
|
| [Hardware Compatibility](docs/hardware-compatibility.md) | Tested boards, minimum requirements |
|
||||||
|
|
||||||
## 🤝 Contribute & Roadmap
|
## 🤝 Contribute & Roadmap
|
||||||
|
|
|
||||||
86
docs/api.md
Normal file
86
docs/api.md
Normal file
|
|
@ -0,0 +1,86 @@
|
||||||
|
# 🌐 Gateway HTTP API Reference
|
||||||
|
|
||||||
|
The PicoClaw gateway provides several HTTP endpoints for health monitoring, management, and direct chat interaction.
|
||||||
|
|
||||||
|
By default, the gateway listens on `127.0.0.1:18790`.
|
||||||
|
|
||||||
|
## 💬 Chat API
|
||||||
|
|
||||||
|
The `/chat` (and alias `/cgat`) endpoint allows you to interact with the PicoClaw agent via a simple HTTP interface. This API is designed to be **asynchronous** to avoid timeouts during long-running LLM tasks or tool executions.
|
||||||
|
|
||||||
|
### 1. Initiate a Chat Session (POST)
|
||||||
|
|
||||||
|
Start a new chat request.
|
||||||
|
|
||||||
|
**Endpoint:** `POST /chat` (or `POST /cgat`)
|
||||||
|
**Content-Type:** `application/json`
|
||||||
|
|
||||||
|
**Request Body:**
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"message": "What is the capital of France?",
|
||||||
|
"session_id": "optional-custom-id"
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Response (202 Accepted):**
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"session_id": "chat-1711352400000",
|
||||||
|
"status": "pending"
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### 2. Poll for Results (GET)
|
||||||
|
|
||||||
|
Retrieve the status and response of a previously initiated session.
|
||||||
|
|
||||||
|
**Endpoint:** `GET /chat?session_id=<ID>` (or `GET /cgat?session_id=<ID>`)
|
||||||
|
|
||||||
|
**Possible Responses:**
|
||||||
|
|
||||||
|
* **Still processing (200 OK):**
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"session_id": "chat-123",
|
||||||
|
"status": "pending"
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
* **Completed (200 OK):**
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"session_id": "chat-123",
|
||||||
|
"status": "completed",
|
||||||
|
"response": "The capital of France is Paris."
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
* **Error (500 Internal Server Error):**
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"session_id": "chat-123",
|
||||||
|
"status": "error",
|
||||||
|
"error": "LLM call failed: context deadline exceeded"
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### 💾 Data Persistence & Cleanup
|
||||||
|
- **Expiry:** Completed or failed results are kept for **1 hour**. Pending sessions are kept for **2 hours**.
|
||||||
|
- **In-Memory:** Results are stored in memory and are lost if the gateway process is restarted.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 🛠️ Management Endpoints
|
||||||
|
|
||||||
|
### Health Check
|
||||||
|
`GET /health`
|
||||||
|
Returns `OK` (200) if the server is running. Used for basic uptime monitoring.
|
||||||
|
|
||||||
|
### Readiness Check
|
||||||
|
`GET /ready`
|
||||||
|
Returns `OK` (200) once the gateway and all enabled channels have successfully initialized.
|
||||||
|
|
||||||
|
### Configuration Reload
|
||||||
|
`POST /reload`
|
||||||
|
Triggers a hot-reload of the `.picoclaw/config.json` file without restarting the process.
|
||||||
|
|
@ -19,9 +19,19 @@ type ChatRequest struct {
|
||||||
SessionID string `json:"session_id,omitempty"`
|
SessionID string `json:"session_id,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// ChatResponse is the JSON response from POST /chat.
|
// ChatResponse is the JSON response from /chat.
|
||||||
type ChatResponse struct {
|
type ChatResponse struct {
|
||||||
Response string `json:"response"`
|
Response string `json:"response,omitempty"`
|
||||||
|
SessionID string `json:"session_id,omitempty"`
|
||||||
|
Status string `json:"status,omitempty"`
|
||||||
|
Error string `json:"error,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type chatStatus struct {
|
||||||
|
Response string
|
||||||
|
Error error
|
||||||
|
Done bool
|
||||||
|
CreatedAt time.Time
|
||||||
}
|
}
|
||||||
|
|
||||||
type Server struct {
|
type Server struct {
|
||||||
|
|
@ -33,6 +43,8 @@ type Server struct {
|
||||||
reloadFunc func() error
|
reloadFunc func() error
|
||||||
chatFunc func(ctx context.Context, message, sessionID string) (string, error)
|
chatFunc func(ctx context.Context, message, sessionID string) (string, error)
|
||||||
apiKey string
|
apiKey string
|
||||||
|
chatResults map[string]*chatStatus
|
||||||
|
chatResultsMu sync.RWMutex
|
||||||
}
|
}
|
||||||
|
|
||||||
type Check struct {
|
type Check struct {
|
||||||
|
|
@ -55,12 +67,17 @@ func NewServer(host string, port int) *Server {
|
||||||
ready: false,
|
ready: false,
|
||||||
checks: make(map[string]Check),
|
checks: make(map[string]Check),
|
||||||
startTime: time.Now(),
|
startTime: time.Now(),
|
||||||
|
chatResults: make(map[string]*chatStatus),
|
||||||
}
|
}
|
||||||
|
|
||||||
mux.HandleFunc("/health", s.healthHandler)
|
mux.HandleFunc("/health", s.healthHandler)
|
||||||
mux.HandleFunc("/ready", s.readyHandler)
|
mux.HandleFunc("/ready", s.readyHandler)
|
||||||
mux.HandleFunc("/reload", s.reloadHandler)
|
mux.HandleFunc("/reload", s.reloadHandler)
|
||||||
mux.HandleFunc("/chat", s.chatHandler)
|
mux.HandleFunc("/chat", s.chatHandler)
|
||||||
|
mux.HandleFunc("/cgat", s.chatHandler)
|
||||||
|
|
||||||
|
// Start task cleanup goroutine
|
||||||
|
go s.taskCleanupLoop()
|
||||||
|
|
||||||
addr := fmt.Sprintf("%s:%d", host, port)
|
addr := fmt.Sprintf("%s:%d", host, port)
|
||||||
s.server = &http.Server{
|
s.server = &http.Server{
|
||||||
|
|
@ -254,29 +271,40 @@ func (s *Server) RegisterOnMux(mux *http.ServeMux) {
|
||||||
mux.HandleFunc("/ready", s.readyHandler)
|
mux.HandleFunc("/ready", s.readyHandler)
|
||||||
mux.HandleFunc("/reload", s.reloadHandler)
|
mux.HandleFunc("/reload", s.reloadHandler)
|
||||||
mux.HandleFunc("/chat", s.chatHandler)
|
mux.HandleFunc("/chat", s.chatHandler)
|
||||||
|
mux.HandleFunc("/cgat", s.chatHandler)
|
||||||
mux.HandleFunc("/v1/chat/completions", func(w http.ResponseWriter, r *http.Request) {
|
mux.HandleFunc("/v1/chat/completions", func(w http.ResponseWriter, r *http.Request) {
|
||||||
logger.Error("GATEWAY IS HITTING ITSELF FOR LLM CALLS!")
|
logger.Error("GATEWAY IS HITTING ITSELF FOR LLM CALLS!")
|
||||||
http.Error(w, "GATEWAY LOOP DETECTION", http.StatusLoopDetected)
|
http.Error(w, "GATEWAY LOOP DETECTION", http.StatusLoopDetected)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// chatHandler handles POST /chat — a synchronous HTTP chat API.
|
// chatHandler handles POST /chat (initiate async) and GET /chat (poll for result).
|
||||||
// Request body: {"message": "...", "session_id": "..." (optional)}
|
// POST body: {"message": "...", "session_id": "..." (optional)}
|
||||||
// Response body: {"response": "..."}
|
// POST response: {"session_id": "...", "status": "pending"}
|
||||||
|
// GET query: ?session_id=...
|
||||||
|
// GET response: {"response": "...", "status": "completed"}
|
||||||
func (s *Server) chatHandler(w http.ResponseWriter, r *http.Request) {
|
func (s *Server) chatHandler(w http.ResponseWriter, r *http.Request) {
|
||||||
if !s.verifyAPIKey(r) {
|
if !s.verifyAPIKey(r) {
|
||||||
w.Header().Set("Content-Type", "application/json")
|
w.Header().Set("Content-Type", "application/json")
|
||||||
w.WriteHeader(http.StatusUnauthorized)
|
w.WriteHeader(http.StatusUnauthorized)
|
||||||
json.NewEncoder(w).Encode(map[string]string{"error": "unauthorized"})
|
json.NewEncoder(w).Encode(ChatResponse{Error: "unauthorized"})
|
||||||
return
|
|
||||||
}
|
|
||||||
if r.Method != http.MethodPost {
|
|
||||||
w.Header().Set("Content-Type", "application/json")
|
|
||||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
|
||||||
json.NewEncoder(w).Encode(map[string]string{"error": "method not allowed, use POST"})
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if r.Method == http.MethodPost {
|
||||||
|
s.handlePostChat(w, r)
|
||||||
|
return
|
||||||
|
} else if r.Method == http.MethodGet {
|
||||||
|
s.handleGetChat(w, r)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||||
|
json.NewEncoder(w).Encode(ChatResponse{Error: "method not allowed, use POST or GET"})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) handlePostChat(w http.ResponseWriter, r *http.Request) {
|
||||||
s.mu.RLock()
|
s.mu.RLock()
|
||||||
chatFunc := s.chatFunc
|
chatFunc := s.chatFunc
|
||||||
s.mu.RUnlock()
|
s.mu.RUnlock()
|
||||||
|
|
@ -284,7 +312,7 @@ func (s *Server) chatHandler(w http.ResponseWriter, r *http.Request) {
|
||||||
if chatFunc == nil {
|
if chatFunc == nil {
|
||||||
w.Header().Set("Content-Type", "application/json")
|
w.Header().Set("Content-Type", "application/json")
|
||||||
w.WriteHeader(http.StatusServiceUnavailable)
|
w.WriteHeader(http.StatusServiceUnavailable)
|
||||||
json.NewEncoder(w).Encode(map[string]string{"error": "chat not configured"})
|
json.NewEncoder(w).Encode(ChatResponse{Error: "chat not configured"})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -292,27 +320,130 @@ func (s *Server) chatHandler(w http.ResponseWriter, r *http.Request) {
|
||||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||||
w.Header().Set("Content-Type", "application/json")
|
w.Header().Set("Content-Type", "application/json")
|
||||||
w.WriteHeader(http.StatusBadRequest)
|
w.WriteHeader(http.StatusBadRequest)
|
||||||
json.NewEncoder(w).Encode(map[string]string{"error": "invalid JSON: " + err.Error()})
|
json.NewEncoder(w).Encode(ChatResponse{Error: "invalid JSON: " + err.Error()})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if req.Message == "" {
|
if req.Message == "" {
|
||||||
w.Header().Set("Content-Type", "application/json")
|
w.Header().Set("Content-Type", "application/json")
|
||||||
w.WriteHeader(http.StatusBadRequest)
|
w.WriteHeader(http.StatusBadRequest)
|
||||||
json.NewEncoder(w).Encode(map[string]string{"error": "message field is required"})
|
json.NewEncoder(w).Encode(ChatResponse{Error: "message field is required"})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
reply, err := chatFunc(r.Context(), req.Message, req.SessionID)
|
sessionID := req.SessionID
|
||||||
if err != nil {
|
if sessionID == "" {
|
||||||
|
sessionID = fmt.Sprintf("chat-%d", time.Now().UnixNano())
|
||||||
|
}
|
||||||
|
|
||||||
|
// Initialize status
|
||||||
|
s.chatResultsMu.Lock()
|
||||||
|
s.chatResults[sessionID] = &chatStatus{
|
||||||
|
CreatedAt: time.Now(),
|
||||||
|
}
|
||||||
|
s.chatResultsMu.Unlock()
|
||||||
|
|
||||||
|
// Start processing in background
|
||||||
|
go func() {
|
||||||
|
// Use a long-running context for the chat call, but don't bind to r.Context()
|
||||||
|
// which will be cancelled when this request finishes.
|
||||||
|
ctx := context.Background()
|
||||||
|
logger.Debugf("Starting async chat for session %s", sessionID)
|
||||||
|
reply, err := chatFunc(ctx, req.Message, sessionID)
|
||||||
|
|
||||||
|
s.chatResultsMu.Lock()
|
||||||
|
defer s.chatResultsMu.Unlock()
|
||||||
|
if result, ok := s.chatResults[sessionID]; ok {
|
||||||
|
result.Response = reply
|
||||||
|
result.Error = err
|
||||||
|
result.Done = true
|
||||||
|
logger.Debugf("Finished async chat for session %s (err=%v)", sessionID, err)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
w.WriteHeader(http.StatusAccepted)
|
||||||
|
json.NewEncoder(w).Encode(ChatResponse{
|
||||||
|
SessionID: sessionID,
|
||||||
|
Status: "pending",
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) handleGetChat(w http.ResponseWriter, r *http.Request) {
|
||||||
|
sessionID := r.URL.Query().Get("session_id")
|
||||||
|
if sessionID == "" {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
w.WriteHeader(http.StatusBadRequest)
|
||||||
|
json.NewEncoder(w).Encode(ChatResponse{Error: "session_id query parameter is required"})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
s.chatResultsMu.RLock()
|
||||||
|
result, ok := s.chatResults[sessionID]
|
||||||
|
if !ok {
|
||||||
|
s.chatResultsMu.RUnlock()
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
w.WriteHeader(http.StatusNotFound)
|
||||||
|
json.NewEncoder(w).Encode(ChatResponse{Error: "session not found"})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Read fields while holding the lock to avoid race conditions
|
||||||
|
done := result.Done
|
||||||
|
response := result.Response
|
||||||
|
errVal := result.Error
|
||||||
|
s.chatResultsMu.RUnlock()
|
||||||
|
|
||||||
|
if !done {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
json.NewEncoder(w).Encode(ChatResponse{
|
||||||
|
SessionID: sessionID,
|
||||||
|
Status: "pending",
|
||||||
|
})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if errVal != nil {
|
||||||
w.Header().Set("Content-Type", "application/json")
|
w.Header().Set("Content-Type", "application/json")
|
||||||
w.WriteHeader(http.StatusInternalServerError)
|
w.WriteHeader(http.StatusInternalServerError)
|
||||||
json.NewEncoder(w).Encode(map[string]string{"error": err.Error()})
|
json.NewEncoder(w).Encode(ChatResponse{
|
||||||
|
SessionID: sessionID,
|
||||||
|
Status: "error",
|
||||||
|
Error: errVal.Error(),
|
||||||
|
})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
w.Header().Set("Content-Type", "application/json")
|
w.Header().Set("Content-Type", "application/json")
|
||||||
w.WriteHeader(http.StatusOK)
|
w.WriteHeader(http.StatusOK)
|
||||||
json.NewEncoder(w).Encode(ChatResponse{Response: reply})
|
json.NewEncoder(w).Encode(ChatResponse{
|
||||||
|
SessionID: sessionID,
|
||||||
|
Status: "completed",
|
||||||
|
Response: response,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) taskCleanupLoop() {
|
||||||
|
ticker := time.NewTicker(10 * time.Minute)
|
||||||
|
defer ticker.Stop()
|
||||||
|
|
||||||
|
for range ticker.C {
|
||||||
|
s.chatResultsMu.Lock()
|
||||||
|
now := time.Now()
|
||||||
|
for id, status := range s.chatResults {
|
||||||
|
// Keep pending tasks for 2 hours, completed/error for 1 hour
|
||||||
|
expiry := time.Hour
|
||||||
|
if !status.Done {
|
||||||
|
expiry = 2 * time.Hour
|
||||||
|
}
|
||||||
|
|
||||||
|
if now.Sub(status.CreatedAt) > expiry {
|
||||||
|
delete(s.chatResults, id)
|
||||||
|
logger.Debugf("Cleaned up expired chat session %s", id)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
s.chatResultsMu.Unlock()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func statusString(ok bool) string {
|
func statusString(ok bool) string {
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue