alex/LocalAGI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add one-time process, attach to UI the mcp server json configuration

Browse Source 
Signed-off-by: mudler <mudler@localai.io>
This commit is contained in:
mudler
2025-04-23 18:28:54 +02:00
parent 0e1106eaf5
commit bd903a3f33
4 changed files with 302 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
pkg/stdio/client.go
View File

@@ -6,6 +6,7 @@ import (
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"net/url"
"sync"
@@ -33,6 +34,8 @@ func NewClient(baseURL string) *Client {
// CreateProcess starts a new process in a group
func (c *Client) CreateProcess(ctx context.Context, command string, args []string, env []string, groupID string) (*Process, error) {
log.Printf("Creating process: command=%s, args=%v, groupID=%s", command, args, groupID)
req := struct {
Command string `json:"command"`
Args []string `json:"args"`
@@ -50,23 +53,28 @@ func (c *Client) CreateProcess(ctx context.Context, command string, args []strin
return nil, fmt.Errorf("failed to marshal request: %w", err)
}
resp, err := http.Post(
fmt.Sprintf("%s/processes", c.baseURL),
"application/json",
bytes.NewReader(reqBody),
)
url := fmt.Sprintf("%s/processes", c.baseURL)
log.Printf("Sending POST request to %s", url)
resp, err := http.Post(url, "application/json", bytes.NewReader(reqBody))
if err != nil {
return nil, fmt.Errorf("failed to start process: %w", err)
}
defer resp.Body.Close()
log.Printf("Received response with status: %d", resp.StatusCode)
var result struct {
ID string `json:"id"`
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return nil, fmt.Errorf("failed to decode response: %w", err)
body, _ := io.ReadAll(resp.Body)
return nil, fmt.Errorf("failed to decode response: %w. body: %s", err, string(body))
}
log.Printf("Successfully created process with ID: %s", result.ID)
process := &Process{
ID: result.ID,
GroupID: groupID,
@@ -193,23 +201,35 @@ func (c *Client) ListGroups() []string {
// GetProcessIO returns io.Reader and io.Writer for a process
func (c *Client) GetProcessIO(id string) (io.Reader, io.Writer, error) {
log.Printf("Getting IO for process: %s", id)
process, err := c.GetProcess(id)
if err != nil {
return nil, nil, err
}
// Parse the base URL to get the host
baseURL, err := url.Parse(c.baseURL)
if err != nil {
return nil, nil, fmt.Errorf("failed to parse base URL: %w", err)
}
// Connect to WebSocket
u := url.URL{
Scheme: "ws",
Host: c.baseURL,
Host: baseURL.Host,
Path: fmt.Sprintf("/ws/%s", process.ID),
}
log.Printf("Connecting to WebSocket at: %s", u.String())
conn, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
if err != nil {
return nil, nil, fmt.Errorf("failed to connect to WebSocket: %w", err)
}
log.Printf("Successfully connected to WebSocket for process: %s", id)
// Create reader and writer
reader := &websocketReader{conn: conn}
writer := &websocketWriter{conn: conn}
@@ -258,3 +278,46 @@ func (c *Client) Close() error {
return nil
}
// RunProcess executes a command and returns its output
func (c *Client) RunProcess(ctx context.Context, command string, args []string, env []string) (string, error) {
log.Printf("Running one-time process: command=%s, args=%v", command, args)
req := struct {
Command string `json:"command"`
Args []string `json:"args"`
Env []string `json:"env"`
}{
Command: command,
Args: args,
Env: env,
}
reqBody, err := json.Marshal(req)
if err != nil {
return "", fmt.Errorf("failed to marshal request: %w", err)
}
url := fmt.Sprintf("%s/run", c.baseURL)
log.Printf("Sending POST request to %s", url)
resp, err := http.Post(url, "application/json", bytes.NewReader(reqBody))
if err != nil {
return "", fmt.Errorf("failed to execute process: %w", err)
}
defer resp.Body.Close()
log.Printf("Received response with status: %d", resp.StatusCode)
var result struct {
Output string `json:"output"`
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
body, _ := io.ReadAll(resp.Body)
return "", fmt.Errorf("failed to decode response: %w. body: %s", err, string(body))
}
log.Printf("Successfully executed process with output length: %d", len(result.Output))
return result.Output, nil
}

66
pkg/stdio/server.go
View File

@@ -5,6 +5,7 @@ import (
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"os"
"os/exec"
@@ -46,6 +47,8 @@ func NewServer() *Server {
// StartProcess starts a new process and returns its ID
func (s *Server) StartProcess(ctx context.Context, command string, args []string, env []string, groupID string) (string, error) {
log.Printf("Starting process: command=%s, args=%v, groupID=%s", command, args, groupID)
cmd := exec.CommandContext(ctx, command, args...)
if len(env) > 0 {
@@ -88,6 +91,7 @@ func (s *Server) StartProcess(ctx context.Context, command string, args []string
}
s.mu.Unlock()
log.Printf("Successfully started process with ID: %s", process.ID)
return process.ID, nil
}
@@ -201,6 +205,22 @@ func (s *Server) ListProcesses() []*Process {
return processes
}
// RunProcess executes a command and returns its output
func (s *Server) RunProcess(ctx context.Context, command string, args []string, env []string) (string, error) {
cmd := exec.CommandContext(ctx, command, args...)
if len(env) > 0 {
cmd.Env = append(os.Environ(), env...)
}
output, err := cmd.CombinedOutput()
if err != nil {
return string(output), fmt.Errorf("process failed: %w", err)
}
return string(output), nil
}
// Start starts the HTTP server
func (s *Server) Start(addr string) error {
http.HandleFunc("/processes", s.handleProcesses)
@@ -208,11 +228,14 @@ func (s *Server) Start(addr string) error {
http.HandleFunc("/ws/", s.handleWebSocket)
http.HandleFunc("/groups", s.handleGroups)
http.HandleFunc("/groups/", s.handleGroup)
http.HandleFunc("/run", s.handleRun)
return http.ListenAndServe(addr, nil)
}
func (s *Server) handleProcesses(w http.ResponseWriter, r *http.Request) {
log.Printf("Handling /processes request: method=%s", r.Method)
switch r.Method {
case http.MethodGet:
processes := s.ListProcesses()
@@ -266,6 +289,8 @@ func (s *Server) handleProcess(w http.ResponseWriter, r *http.Request) {
func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) {
id := r.URL.Path[len("/ws/"):]
log.Printf("Handling WebSocket connection for process: %s", id)
process, err := s.GetProcess(id)
if err != nil {
http.Error(w, err.Error(), http.StatusNotFound)
@@ -279,11 +304,14 @@ func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) {
}
defer conn.Close()
log.Printf("WebSocket connection established for process: %s", id)
// Handle stdin
go func() {
for {
_, message, err := conn.ReadMessage()
if err != nil {
log.Printf("WebSocket stdin read error for process %s: %v", id, err)
return
}
process.Stdin.Write(message)
@@ -294,9 +322,9 @@ func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) {
go func() {
buf := make([]byte, 1024)
for {
n, err := process.Stdout.Read(buf)
if err != nil {
log.Printf("WebSocket stdout read error for process %s: %v", id, err)
return
}
conn.WriteMessage(websocket.TextMessage, buf[:n])
@@ -309,6 +337,7 @@ func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) {
for {
n, err := process.Stderr.Read(buf)
if err != nil {
log.Printf("WebSocket stderr read error for process %s: %v", id, err)
return
}
conn.WriteMessage(websocket.TextMessage, buf[:n])
@@ -317,6 +346,7 @@ func (s *Server) handleWebSocket(w http.ResponseWriter, r *http.Request) {
// Wait for process to exit
process.Cmd.Wait()
log.Printf("Process %s exited", id)
}
// Add new handlers for group management
@@ -351,3 +381,37 @@ func (s *Server) handleGroup(w http.ResponseWriter, r *http.Request) {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
}
}
func (s *Server) handleRun(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
log.Printf("Handling /run request")
var req struct {
Command string `json:"command"`
Args []string `json:"args"`
Env []string `json:"env"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
log.Printf("Executing one-time process: command=%s, args=%v", req.Command, req.Args)
output, err := s.RunProcess(r.Context(), req.Command, req.Args, req.Env)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
log.Printf("One-time process completed with output length: %d", len(output))
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]string{
"output": output,
})
}