Lot 20 : serveurs MCP tenus par la façade (LOKI_MCP=rust)

La façade tient les serveurs MCP quand LOKI_MCP=rust (expérimental ; le
Go reste le défaut) : configuration (state/mcp), catalogue embarqué (le
même fichier que le Go, copié dans l'étape Rust de l'image), état et
routes /api/mcp*. Un client MCP en Rust (mcp/client.rs) parle JSON-RPC
sur stdio et en Streamable HTTP (réponses JSON ou SSE, session suivie,
en-têtes configurés, ping du serveur servi). Le pool suit mcp_client.go :
session gardée entre les tours, connexions parallèles et partagées,
session invalidée pendant sa connexion jamais republiée, délais npx/uvx,
appel borné avec une reconnexion, sortie plafonnée à la fenêtre.

Un seul pool, celui de la façade : le Go, qui exécute encore les outils
de l'agent, lui demande les outils MCP, leurs appels et le
pré-chauffage (/internal/facade/v1/mcp/*, facade_mcp.go).

Parité : parity_mcp_test.go tire du Go les fonctions pures, seize
mutations de configuration, et un vrai serveur du SDK MCP Go (onze
outils) : ce qu'il met sur le fil et ce que le client Go en tire (état,
outils annoncés, appels). tests/mcp.rs le rejoue contre un faux serveur
qui resert ces réponses, et vérifie stdio, reconnexion, routes, catalogue
et session invalidée de bout en bout ; TestFacadeMCP côté Go.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016tcGfJWefW5kUGoC2wL8cv
This commit is contained in:
Claude committed 2026-10-15 19:56:11 +00:00
1 parent 3fc36a3acc
commit 090f15615e
20 files changed
+3670 -2

No files matched your search

+2
View File
@@ -53,6 +53,8 @@ WORKDIR /src
# L'UI est embarquée par build.rs depuis la même source que le go:embed : les
# deux binaires d'une image servent toujours la même interface.
COPY internal/loki/ui/ internal/loki/ui/
# Catalogue MCP embarqué par loki-rs (lot 20), le même que le Go.
COPY internal/loki/mcp_catalog.json internal/loki/mcp_catalog.json
COPY rust/ rust/
RUN cd rust && cargo build --release --locked \
&& mkdir -p /out && cp target/release/loki-rs /out/loki-rs
+1
View File
@@ -236,6 +236,7 @@ redémarrages (volume `/data`). Variables d'environnement du conteneur :
| `LOKI_MEM` | `rust` : la mémoire (éditeur des pages, index, déplacement entre projets, trackers, mode mémoire, interrupteurs du mode agent) est tenue par la façade Rust ; le chiffrement reste au Go (expérimental, migration en cours) | Go |
| `LOKI_SERVE` | `rust` : la ligne de commande de llama-server est composée par la façade Rust (`loki-rs engine serve`) au lieu de `loki serve` ; Strata et l'optimiseur restent au Go (expérimental, migration en cours) | Go |
| `LOKI_PRESETS` | `rust` : les presets (liste, éditeur et aperçus, suppression, placement auto, presets externes, bascule) sont tenus par la façade Rust ; l'application au moteur reste au Go (expérimental, migration en cours) | Go |
| `LOKI_MCP` | `rust` : les serveurs MCP — configuration, catalogue, état, pool de sessions stdio et HTTP — sont tenus par la façade Rust ; le Go lui demande les outils MCP de l'agent et leurs appels (expérimental, migration en cours) | Go |
| `LOKI_PROJECTS` | `rust` : le hub des projets (liste et comptes, création, renommage, description, suppression, bascule) est tenu par la façade Rust ; les discussions restent à qui tient la conversation (expérimental, migration en cours) | Go |
| `LOKI_SYSTEM` | `rust` : les interrupteurs système — accès internet et Crawl4AI, pilotage du navigateur, clé de complétion, écoute réseau du moteur — sont tenus par la façade Rust ; le navigateur piloté reste au Go (expérimental, migration en cours) | Go |
| `LOKI_ENGINE` | `rust` : le pilotage du moteur (démarrage, arrêt, redémarrage, déchargement et rechargement de la VRAM) est tenu par la façade Rust, en conteneur avec son superviseur ; le verrou de l'optimiseur reste au Go (expérimental, migration en cours) | Go |
@@ -0,0 +1,93 @@
# Lot 20 — serveurs MCP tenus par la façade
## Contexte
Le Go tenait les serveurs MCP (Model Context Protocol) :
- `mcp_config.go` : la configuration, dans `state/mcp` ;
- `mcp_catalog.go` : le catalogue embarqué ;
- `mcp_client.go` : le pool de sessions, avec le SDK MCP Go ;
- les routes `/api/mcp*` de `web_api.go`.
Le pool sert deux usages : l'état affiché dans l'interface, et les outils
`mcp__<serveur>__<outil>` annoncés au modèle et exécutés pour l'agent.
Lot 20 : la façade tient tout cela quand `LOKI_MCP=rust` (`mcp/`,
expérimental). Sans la variable, rien ne change.
| Route | Rôle |
|---|---|
| `GET /api/mcp` | état de chaque serveur, connexion tentée pour les activés (en parallèle) |
| `GET /api/mcp/catalog` | catalogue embarqué (le même fichier que le Go), `{workspace}` résolu, entrées déjà configurées marquées, lanceur (npx, uvx) présent ou non |
| `POST /api/mcp/save` | ajoute ou remplace (actif par défaut) ; nom contrôlé, un seul transport |
| `POST /api/mcp/delete`, `/api/mcp/toggle`, `/api/mcp/tool` | retire, active ou éteint, masque ou démasque un outil |
| `POST /api/mcp/test` | reconnexion forcée |
## Client MCP en Rust
`mcp/client.rs` parle JSON-RPC 2.0 sur deux transports, comme le SDK Go :
- stdio : un processus local, un message par ligne ;
- Streamable HTTP : un POST par message, réponse en JSON ou en flux SSE,
session suivie par `Mcp-Session-Id`, en-têtes configurés ajoutés.
La session s'ouvre par `initialize` puis `notifications/initialized`. Elle
sert `tools/list` (première page, comme le Go) et `tools/call`. Un `ping` du
serveur reçoit sa réponse.
Le pool (`mcp/mod.rs`) suit `mcp_client.go` :
- une session gardée d'un tour à l'autre ;
- une entrée publiée avant la connexion, pour que deux appelants partagent
la même tentative ;
- une session invalidée pendant sa connexion n'est jamais republiée ;
- délai de 3 min pour npx et uvx (premier téléchargement), 20 s sinon ;
- un appel borné à 120 s, une reconnexion si la session est morte ;
- la sortie plafonnée à la fenêtre de contexte (12 000 au moins), une erreur
d'outil signalée, une sortie vide rendue « [ok] (aucune sortie) ».
## Le Go
Un seul pool, celui de la façade. Le Go, qui exécute encore les outils de
l'agent, lui demande les outils à annoncer, leurs appels et le
pré-chauffage (`/internal/facade/v1/mcp/{tools,call,prewarm}`,
`facade_mcp.go`), quelle que soit la conversation qui tourne. Son propre
pool reste vide.
L'image copie `internal/loki/mcp_catalog.json` dans l'étape de compilation
Rust : un seul catalogue pour les deux binaires.
## Écarts assumés
- Les erreurs de transport (connexion refusée, flux coupé) ont les messages
du client Rust. Une erreur JSON-RPC et un binaire introuvable ont ceux du
Go.
- Un contenu « autre » (lien de ressource, ressource intégrée) est rendu en
JSON dans l'ordre des champs du SDK. Les annotations éventuelles sont dans
l'ordre alphabétique.
## Parité
`internal/loki/parity_mcp_test.go` tire du Go :
- les fonctions pures : noms exposés, schémas, libellés d'appel, serveur
d'origine d'un outil, ligne du préambule, délais et leurs messages,
aplatissement de neuf résultats ;
- la lecture de la configuration et seize mutations avec leurs erreurs et la
base après chacune ;
- ce qu'un vrai serveur du SDK MCP Go met sur le fil en Streamable HTTP :
onze outils, et le résultat de chacun ;
- ce que le client Go en tire : état de quatre serveurs (joignable,
binaire introuvable, désactivé, mal configuré), outils annoncés (un
masqué), appels.
`tests/mcp.rs` rejoue les fonctions et la configuration. Un faux serveur
resert ce que le serveur du SDK a répondu (appels en SSE) : l'état, les
outils et les appels tirés par la façade doivent être ceux du Go, sur une
seule session. Sont aussi vérifiés de bout en bout :
- le transport stdio : environnement, ping du serveur, serveur mort en plein
appel puis reconnexion ;
- les routes de configuration, en-têtes compris ;
- le catalogue ;
- une session invalidée pendant sa connexion, sans processus laissé
derrière.
Une mutation de chaque règle fait échouer les tests. Côté Go,
`TestFacadeMCP` vérifie le relais, les outils, les appels, le
pré-chauffage et une façade injoignable.
+3 -1
View File
@@ -150,7 +150,7 @@ var facadeChatRoutes = map[string]bool{
// de la discussion (facade_files.go), les notifications (facade_push.go), la
// mémoire (facade_mem.go), les presets (facade_presets.go), le pilotage du
// moteur (facade_engine.go), les interrupteurs système (facade_system.go), les
// projets (facade_projects.go).
// projets (facade_projects.go), les serveurs MCP (facade_mcp.go).
// Posé APRÈS requireWebAuth : seule une requête authentifiée ici part, avec le
// jeton interne qui en tient lieu là-bas.
func withFacadeChat(path string, h http.HandlerFunc) http.HandlerFunc {
@@ -177,6 +177,8 @@ func withFacadeChat(path string, h http.HandlerFunc) http.HandlerFunc {
held, unavailable = facadeSystem, "réglages système indisponibles : "
case facadeProjectsRoutes[path]:
held, unavailable = facadeProjects, "projets indisponibles : "
case facadeMCPRoutes[path]:
held, unavailable = facadeMCP, "serveurs MCP indisponibles : "
default:
return h
}
+70
View File
@@ -10,6 +10,7 @@ import (
"strings"
"sync"
"testing"
"time"
)
// fausseFacade : enregistre ce qui lui arrive ; /api/chat rend un flux.
@@ -835,3 +836,72 @@ func TestFacadeProjets(t *testing.T) {
t.Fatalf("corps illisible : %d", c)
}
}
// Les serveurs MCP (LOKI_MCP=rust, lot 20) : routes relayées, et le Go
// demande à la façade les outils MCP et leurs appels (un seul pool).
func TestFacadeMCP(t *testing.T) {
testHome(t)
var calls []string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get(internalTokenHeader) != jetonTest {
w.WriteHeader(403)
return
}
b, _ := io.ReadAll(r.Body)
calls = append(calls, r.URL.Path+" "+string(b))
switch r.URL.Path {
case "/internal/facade/v1/mcp/tools":
io.WriteString(w, `{"tools":[{"type":"function","function":{"name":"mcp__a__t","description":"[MCP: a] d","parameters":{"type":"object"}}}]}`)
case "/internal/facade/v1/mcp/call":
io.WriteString(w, `{"result":"fait"}`)
case "/internal/facade/v1/mcp/prewarm":
io.WriteString(w, `{"ok":true}`)
default:
io.WriteString(w, `{"ok":true,"de":"facade"}`)
}
}))
defer srv.Close()
t.Setenv(internalTokenEnv, jetonTest)
t.Setenv(facadeAddrEnv, strings.TrimPrefix(srv.URL, "http://"))
local := func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "go") }
t.Setenv(facadeMCPEnv, "")
w := httptest.NewRecorder()
withFacadeChat("/api/mcp", local)(w, httptest.NewRequest("GET", "/api/mcp", nil))
if w.Body.String() != "go" || len(calls) != 0 {
t.Fatalf("sans LOKI_MCP : %q", w.Body.String())
}
t.Setenv(facadeMCPEnv, "rust")
for p := range facadeMCPRoutes {
w := httptest.NewRecorder()
withFacadeChat(p, local)(w, httptest.NewRequest("POST", p, strings.NewReader("{}")))
if !strings.Contains(w.Body.String(), "facade") {
t.Fatalf("%s non relayé : %q", p, w.Body.String())
}
}
tools := mcpTools()
if len(tools) != 1 || tools[0].Function.Name != "mcp__a__t" || mcpPromptLine(tools) == "" {
t.Fatalf("outils : %+v", tools)
}
if got := mcpCall("mcp__a__t", map[string]any{"x": 1}); got != "fait" {
t.Fatalf("appel : %q", got)
}
if got := mcpCall("mcp__a__t", nil); got != "fait" || !strings.HasSuffix(calls[len(calls)-1], `{"args":{},"name":"mcp__a__t"}`) {
t.Fatalf("appel sans arguments : %q (%v)", got, calls)
}
MCPPrewarm()
for i := 0; i < 50 && !strings.HasPrefix(calls[len(calls)-1], "/internal/facade/v1/mcp/prewarm"); i++ {
time.Sleep(20 * time.Millisecond)
}
if !strings.HasPrefix(calls[len(calls)-1], "/internal/facade/v1/mcp/prewarm") {
t.Fatalf("pré-chauffage non demandé : %v", calls)
}
// Façade injoignable : une erreur lisible, pas de panique.
t.Setenv(facadeAddrEnv, "127.0.0.1:1")
if got := mcpCall("mcp__a__t", nil); !strings.HasPrefix(got, "[erreur MCP] façade injoignable") {
t.Fatalf("façade injoignable : %q", got)
}
if tools := mcpTools(); tools != nil {
t.Fatalf("outils sans façade : %v", tools)
}
}
+105
View File
@@ -0,0 +1,105 @@
package loki
import (
"bytes"
"encoding/json"
"fmt"
"net/http"
"os"
"strings"
"time"
)
// facade_mcp.go — serveurs MCP tenus par la façade Rust (LOKI_MCP=rust, lot 20
// de la migration ; expérimental, le Go reste le défaut) : configuration,
// catalogue, état et pool de sessions (rust/loki-rs/src/mcp).
//
// Un seul pool : celui de la façade. Le Go, qui exécute encore les outils de
// l'agent, lui demande les outils MCP à annoncer au modèle et leurs appels
// (/internal/facade/v1/mcp/*), quelle que soit la conversation qui tourne ;
// son propre pool reste vide.
const facadeMCPEnv = "LOKI_MCP"
// facadeMCP : la façade tient-elle les serveurs MCP ?
func facadeMCP() bool {
return strings.EqualFold(strings.TrimSpace(os.Getenv(facadeMCPEnv)), "rust") &&
facadeAddr() != "" && internalToken() != ""
}
// facadeMCPRoutes : les routes que la façade sert alors (mcp/mod.rs).
var facadeMCPRoutes = map[string]bool{
"/api/mcp": true,
"/api/mcp/catalog": true,
"/api/mcp/save": true,
"/api/mcp/delete": true,
"/api/mcp/toggle": true,
"/api/mcp/tool": true,
"/api/mcp/test": true,
}
// facadeMCPJSON appelle l'API interne MCP de la façade et décode sa réponse.
// Délai large : un appel d'outil MCP peut durer (mcpCallTimeout), une
// première connexion npx/uvx aussi (mcpConnectTimeoutCold).
func facadeMCPJSON(path string, body any, out any) error {
b, err := json.Marshal(body)
if err != nil {
return err
}
req, err := http.NewRequest(http.MethodPost, "http://"+facadeAddr()+path, bytes.NewReader(b))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set(internalTokenHeader, internalToken())
resp, err := (&http.Client{Timeout: mcpConnectTimeoutCold + mcpCallTimeout}).Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
return fmt.Errorf("façade : %s", resp.Status)
}
return json.NewDecoder(resp.Body).Decode(out)
}
// facadeMCPTools : les outils MCP à annoncer, tenus par la façade.
func facadeMCPTools() []Tool {
var out struct {
Tools []Tool `json:"tools"`
}
if err := facadeMCPJSON("/internal/facade/v1/mcp/tools", map[string]any{}, &out); err != nil {
fmt.Fprintf(os.Stderr, "[mcp] outils demandés à la façade : %v\n", err)
return nil
}
return out.Tools
}
// facadeMCPCall : un outil MCP exécuté par la façade.
func facadeMCPCall(name string, args map[string]any) string {
if args == nil {
args = map[string]any{}
}
var out struct {
Result string `json:"result"`
}
if err := facadeMCPJSON("/internal/facade/v1/mcp/call", map[string]any{"name": name, "args": args}, &out); err != nil {
return "[erreur MCP] façade injoignable : " + err.Error()
}
return out.Result
}
// facadeMCPPrewarm : la façade ouvre ses sessions en tâche de fond.
func facadeMCPPrewarm() {
go func() {
var out map[string]any
// La façade démarre le Go puis attend qu'il réponde : elle peut ne pas
// écouter encore au tout premier instant.
for i := 0; i < 10; i++ {
if facadeMCPJSON("/internal/facade/v1/mcp/prewarm", map[string]any{}, &out) == nil {
return
}
time.Sleep(time.Second)
}
}()
}
+10
View File
@@ -256,6 +256,10 @@ func mcpEnsureAll(servers map[string]MCPServerConfig) {
// qui donnait l'impression d'une IA très lente à démarrer. Sans mode agent les
// outils MCP ne sont de toute façon pas exposés : inutile de lancer les process.
func MCPPrewarm() {
if facadeMCP() { // le pool est celui de la façade (facade_mcp.go)
facadeMCPPrewarm()
return
}
if !agentEnabled() {
return
}
@@ -282,6 +286,9 @@ func mcpExposedName(server, tool string) string {
// mcpTools connecte les serveurs activés et renvoie leurs outils, prêts à être
// annoncés au modèle. Réservé au mode agent (appelé par EnabledTools).
func mcpTools() []Tool {
if facadeMCP() { // le pool est celui de la façade (facade_mcp.go)
return facadeMCPTools()
}
servers, err := LoadMCPConfig()
if err != nil || len(servers) == 0 {
return nil
@@ -368,6 +375,9 @@ func isMCPTool(name string) bool {
// mcpCall exécute un outil MCP à partir de son nom exposé et renvoie une chaîne
// résultat (texte aplati), cappée. Reconnecte une fois si la session est morte.
func mcpCall(name string, args map[string]any) string {
if facadeMCP() { // le pool est celui de la façade (facade_mcp.go)
return facadeMCPCall(name, args)
}
mcpMgr.mu.Lock()
ref, ok := mcpMgr.registry[name]
mcpMgr.mu.Unlock()
+338
View File
@@ -0,0 +1,338 @@
//go:build linux
package loki
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"testing"
mcpsdk "github.com/modelcontextprotocol/go-sdk/mcp"
)
// parity_mcp_test.go — journal de référence des serveurs MCP (lot 20 de la
// migration Rust) : fonctions pures du client (noms exposés, schémas,
// libellés, ligne du préambule, délais, aplatissement des résultats), lecture
// et mutations de la configuration, puis un vrai serveur du SDK MCP Go servi
// en Streamable HTTP : ce qu'il annonce et renvoie (tel que sur le fil), et ce
// que le client Go en tire (état, outils annoncés au modèle, appels). Rangé
// dans rust/loki-rs/tests/parity_mcp/mcp.json ; tests/mcp.rs le rejoue contre
// un faux serveur qui sert les mêmes réponses.
//
// LOKI_PARITY_UPDATE=1 go test ./internal/loki -run TestPariteMCP
const parityMCPFile = "../../rust/loki-rs/tests/parity_mcp/mcp.json"
type parityMCPMutation struct {
Op string `json:"op"`
Name string `json:"name"`
Tool string `json:"tool,omitempty"`
On bool `json:"on,omitempty"`
Cfg string `json:"cfg,omitempty"` // MCPServerConfig en JSON
Error string `json:"error"`
State string `json:"state"` // state/mcp après l'opération
}
func TestPariteMCP(t *testing.T) {
testHome(t)
t.Cleanup(mcpCloseAll)
t.Setenv(facadeMCPEnv, "")
fx := map[string]any{}
// --- fonctions pures ---
san := map[string]string{}
for _, s := range []string{"alpha", "mon serveur!", "__x__", "a.b/c", "é-t_é", "", "...", "A-B_c9"} {
san[s] = mcpSanitize(s)
}
fx["sanitize"] = san
var schemas []any
for _, raw := range []string{`null`, `{"type":"object","properties":{"a":{"type":"string"}}}`, `true`, `[1]`, `{}`, `"x"`} {
var v any
_ = json.Unmarshal([]byte(raw), &v)
schemas = append(schemas, map[string]any{"in": raw, "out": mcpNormalizeSchema(v)})
}
fx["schemas"] = schemas
var labels []any
for _, raw := range []string{`{}`, `{"path":"/a","query":"q"}`, `{"query":"","url":"http://x"}`, `{"name":"n","command":"c"}`,
`{"x":1,"y":"<b>"}`, `{"long":"` + strings.Repeat("é", 200) + `"}`, `{"path":3,"file":"f"}`,
`{"file":"f","path":"p"}`, `{"moyen":"` + strings.Repeat("a", 150) + `"}`} {
var m map[string]any
_ = json.Unmarshal([]byte(raw), &m)
labels = append(labels, map[string]any{"in": raw, "out": mcpArgLabel(m)})
}
fx["labels"] = labels
var servers []any
for _, c := range [][2]string{
{"mcp__alpha__x", "[MCP: alpha] desc"}, {"mcp__a_b__x", "[MCP: a]b] desc"}, {"mcp__a_b_c__x", "[MCP: a] b] c] d"},
{"mcp__x__y", "pas de préfixe"}, {"autre", "[MCP: autre] d"}, {"mcp__b__x", "[MCP: a] d"},
} {
s, ok := mcpToolServer(Tool{Function: ToolFunction{Name: c[0], Description: c[1]}})
servers = append(servers, map[string]any{"name": c[0], "desc": c[1], "server": s, "ok": ok})
}
fx["tool_server"] = servers
var lines []any
for _, set := range [][][2]string{
{},
{{"bash", "shell"}},
{{"mcp__b__x", "[MCP: b] d"}, {"mcp__a__y", "[MCP: a] d"}, {"mcp__a__z", "[MCP: a] e"}},
{{"mcp__a_b__x", "[MCP: a]b] d"}},
} {
var tools []Tool
for _, p := range set {
tools = append(tools, Tool{Function: ToolFunction{Name: p[0], Description: p[1]}})
}
lines = append(lines, map[string]any{"tools": set, "out": mcpPromptLine(tools)})
}
fx["prompt_line"] = lines
timeouts := map[string]any{}
for _, cmd := range []string{"npx", "/usr/local/bin/NPX", "uvx.exe", "npx.cmd", "node", "", "uvx-x", "C:/bin/uvx.EXE"} {
cfg := MCPServerConfig{Command: cmd}
timeouts[cmd] = map[string]any{"secs": int(mcpTimeoutFor(cfg).Seconds()),
"err": mcpFriendlyErr(context.DeadlineExceeded, cfg).Error()}
}
fx["timeouts"] = timeouts
fx["friendly_other"] = mcpFriendlyErr(errors.New("autre"), MCPServerConfig{Command: "npx"}).Error()
var flat []any
for _, raw := range []string{
`{"content":[{"type":"text","text":" un "},{"type":"text","text":"deux\n"}]}`,
`{"content":[{"type":"image","data":"cG5n","mimeType":"image/png"},{"type":"audio","data":"d2F2","mimeType":"audio/wav"}]}`,
`{"content":[{"type":"resource_link","uri":"file:///a.txt","name":"a.txt","mimeType":"text/plain","size":12,"title":"A <b>"}]}`,
`{"content":[{"type":"resource","resource":{"uri":"file:///b","mimeType":"text/plain","text":"corps"}}]}`,
`{"content":[],"structuredContent":{"b":[1,2],"a":"<x> & y"}}`,
`{"content":[{"type":"text","text":" "}],"structuredContent":{"k":1}}`,
`{"content":[{"type":"text","text":"t"}],"structuredContent":{"k":1}}`,
`{"content":[]}`,
`{"content":[{"type":"text","text":"boum"}],"isError":true}`,
} {
var res mcpsdk.CallToolResult
if err := json.Unmarshal([]byte(raw), &res); err != nil {
t.Fatalf("%s : %v", raw, err)
}
flat = append(flat, map[string]any{"in": raw, "out": flattenMCPContent(&res)})
}
fx["flatten"] = flat
// --- configuration ---
var decoded []any
for _, raw := range []string{
`{"a":{"command":"npx","args":["-y","x"],"env":{"K":"v"}}}`,
`{"b":{"url":"http://h/mcp","headers":{"Authorization":"Bearer t"},"enabled":false,"disabledTools":["t1"]}}`,
`{"c":{"command":"x","enabled":null}}`,
`null`, `{}`,
} {
var m map[string]MCPServerConfig
_ = json.Unmarshal([]byte(raw), &m)
out, _ := json.Marshal(m)
tr := map[string]string{}
for n, c := range m {
tr[n] = c.Transport() + "|" + errString(c.Validate())
}
decoded = append(decoded, map[string]any{"in": raw, "out": string(out), "transports": tr})
}
fx["decode"] = decoded
var muts []parityMCPMutation
cfgOf := func(raw string) MCPServerConfig {
var c MCPServerConfig
_ = json.Unmarshal([]byte(raw), &c)
return c
}
for _, m := range []parityMCPMutation{
{Op: "set", Name: " fs ", Cfg: `{"command":"npx","args":["-y","@x/fs","<w>"],"env":{"A":"&"}}`},
{Op: "set", Name: "", Cfg: `{"command":"x"}`},
{Op: "set", Name: "a__b", Cfg: `{"command":"x"}`},
{Op: "set", Name: "both", Cfg: `{"command":"x","url":"http://u"}`},
{Op: "set", Name: "none", Cfg: `{"enabled":true}`},
{Op: "set", Name: "web", Cfg: `{"url":"http://u/mcp","headers":{"X":"1"},"enabled":false}`},
{Op: "enable", Name: "web", On: true},
{Op: "enable", Name: "nope", On: true},
{Op: "tool", Name: "fs", Tool: "write", On: false},
{Op: "tool", Name: "fs", Tool: "read", On: false},
{Op: "tool", Name: "fs", Tool: "write", On: true},
{Op: "tool", Name: "fs", Tool: "read", On: false},
{Op: "tool", Name: "nope", Tool: "x", On: false},
{Op: "delete", Name: "web"},
{Op: "delete", Name: "web"},
{Op: "delete", Name: " fs"},
} {
var err error
switch m.Op {
case "set":
err = SetMCPServer(m.Name, cfgOf(m.Cfg))
case "enable":
err = SetMCPServerEnabled(m.Name, m.On)
case "tool":
err = SetMCPToolEnabled(m.Name, m.Tool, m.On)
case "delete":
err = DeleteMCPServer(m.Name)
}
m.Error = errString(err)
m.State = getStr(bkState, "mcp")
muts = append(muts, m)
}
fx["mutations"] = muts
// --- un vrai serveur MCP (SDK Go), en Streamable HTTP ---
srvMCP := mcpsdk.NewServer(&mcpsdk.Implementation{Name: "parite", Version: "1"}, nil)
schema := map[string]any{"type": "object", "properties": map[string]any{"text": map[string]any{"type": "string"}}}
text := func(s string) []mcpsdk.Content { return []mcpsdk.Content{&mcpsdk.TextContent{Text: s}} }
tools := []struct {
tool *mcpsdk.Tool
res func(args map[string]any) *mcpsdk.CallToolResult
}{
{&mcpsdk.Tool{Name: "echo", Description: "Répète le texte <tel quel>", InputSchema: schema},
func(a map[string]any) *mcpsdk.CallToolResult {
s, _ := a["text"].(string)
return &mcpsdk.CallToolResult{Content: text(s)}
}},
{&mcpsdk.Tool{Name: "multi", Description: "Deux blocs", InputSchema: map[string]any{"type": "object"}},
func(map[string]any) *mcpsdk.CallToolResult {
return &mcpsdk.CallToolResult{Content: []mcpsdk.Content{&mcpsdk.TextContent{Text: "un"}, &mcpsdk.TextContent{Text: "deux"}}}
}},
{&mcpsdk.Tool{Name: "media", Description: "Image et son", InputSchema: map[string]any{"type": "object"}},
func(map[string]any) *mcpsdk.CallToolResult {
return &mcpsdk.CallToolResult{Content: []mcpsdk.Content{&mcpsdk.ImageContent{Data: []byte("png"), MIMEType: "image/png"},
&mcpsdk.AudioContent{Data: []byte("wav"), MIMEType: "audio/wav"}}}
}},
{&mcpsdk.Tool{Name: "link", Description: "Un lien", InputSchema: map[string]any{"type": "object"}},
func(map[string]any) *mcpsdk.CallToolResult {
sz := int64(12)
return &mcpsdk.CallToolResult{Content: []mcpsdk.Content{&mcpsdk.ResourceLink{URI: "file:///a.txt", Name: "a.txt", Title: "A", MIMEType: "text/plain", Size: &sz}}}
}},
{&mcpsdk.Tool{Name: "struct", Description: "Structuré", InputSchema: map[string]any{"type": "object"}},
func(map[string]any) *mcpsdk.CallToolResult {
return &mcpsdk.CallToolResult{Content: []mcpsdk.Content{}, StructuredContent: map[string]any{"b": []any{1, 2}, "a": "<x> & y"}}
}},
{&mcpsdk.Tool{Name: "fail", Description: "Échoue", InputSchema: map[string]any{"type": "object"}},
func(map[string]any) *mcpsdk.CallToolResult {
return &mcpsdk.CallToolResult{Content: text("boum"), IsError: true}
}},
{&mcpsdk.Tool{Name: "empty", Description: "Rien", InputSchema: map[string]any{"type": "object"}},
func(map[string]any) *mcpsdk.CallToolResult {
return &mcpsdk.CallToolResult{Content: []mcpsdk.Content{}}
}},
{&mcpsdk.Tool{Name: "big", Description: "Long", InputSchema: map[string]any{"type": "object"}},
func(map[string]any) *mcpsdk.CallToolResult {
return &mcpsdk.CallToolResult{Content: text(strings.Repeat("é", 40000))}
}},
{&mcpsdk.Tool{Name: "titled", Title: "Titre seul", InputSchema: map[string]any{"type": "object"}},
func(map[string]any) *mcpsdk.CallToolResult { return &mcpsdk.CallToolResult{Content: text("ok")} }},
{&mcpsdk.Tool{Name: "secret", Description: "Masqué", InputSchema: map[string]any{"type": "object"}},
func(map[string]any) *mcpsdk.CallToolResult { return &mcpsdk.CallToolResult{Content: text("chut")} }},
{&mcpsdk.Tool{Name: "dot.tool", Description: "Nom pointé", InputSchema: map[string]any{"type": "object", "properties": map[string]any{}}},
func(map[string]any) *mcpsdk.CallToolResult { return &mcpsdk.CallToolResult{Content: text("point")} }},
}
for _, tl := range tools {
res := tl.res
srvMCP.AddTool(tl.tool, func(_ context.Context, req *mcpsdk.CallToolRequest) (*mcpsdk.CallToolResult, error) {
var args map[string]any
_ = json.Unmarshal(req.Params.Arguments, &args)
return res(args), nil
})
}
ts := httptest.NewServer(mcpsdk.NewStreamableHTTPHandler(func(*http.Request) *mcpsdk.Server { return srvMCP }, nil))
// Les sessions du pool et du serveur tiennent des connexions ouvertes :
// fermées avant le serveur, sans quoi Close attend indéfiniment.
defer func() {
mcpCloseAll()
for ss := range srvMCP.Sessions() {
_ = ss.Close()
}
ts.CloseClientConnections()
ts.Close()
}()
// Ce que le serveur met sur le fil (le faux serveur Rust le resservira).
cl := mcpsdk.NewClient(&mcpsdk.Implementation{Name: "t", Version: "1"}, nil)
sess, err := cl.Connect(context.Background(), &mcpsdk.StreamableClientTransport{Endpoint: ts.URL}, nil)
if err != nil {
t.Fatal(err)
}
lt, err := sess.ListTools(context.Background(), nil)
if err != nil {
t.Fatal(err)
}
wire := map[string]any{}
b, _ := json.Marshal(lt)
var ltv any
_ = json.Unmarshal(b, &ltv)
wire["tools/list"] = ltv
calls := map[string]any{}
callArgs := map[string]map[string]any{"echo": {"text": "salut <toi>"}}
for _, tl := range tools {
res, err := sess.CallTool(context.Background(), &mcpsdk.CallToolParams{Name: tl.tool.Name, Arguments: callArgs[tl.tool.Name]})
if err != nil {
t.Fatal(err)
}
b, _ := json.Marshal(res)
var v any
_ = json.Unmarshal(b, &v)
calls[tl.tool.Name] = v
}
wire["tools/call"] = calls
_ = sess.Close()
fx["wire"] = wire
// Ce que le client de Loki en tire.
_ = putStr(bkState, "mcp", "")
if err := SetMCPServer("alpha", MCPServerConfig{URL: ts.URL, Enabled: true}); err != nil {
t.Fatal(err)
}
_ = SetMCPToolEnabled("alpha", "secret", false)
_ = SetMCPServer("beta", MCPServerConfig{Command: "loki-mcp-absent-pour-test", Enabled: true})
_ = SetMCPServer("gamma", MCPServerConfig{URL: ts.URL, Enabled: false})
raw := getStr(bkState, "mcp")
raw = strings.Replace(raw, `}}`, `},"delta":{"enabled":true}}`, 1)
_ = putStr(bkState, "mcp", raw)
fx["store"] = strings.ReplaceAll(getStr(bkState, "mcp"), ts.URL, "{{URL}}")
st, err := MCPStatus()
if err != nil {
t.Fatal(err)
}
fx["status"] = st
fx["tools"] = mcpTools()
results := map[string]string{}
for _, n := range []string{"echo", "multi", "media", "link", "struct", "fail", "empty", "big", "titled", "secret", "dot_tool", "nope"} {
name := mcpExposedName("alpha", n)
results[name] = mcpCall(name, callArgs[n])
}
results["mcp__zeta__x"] = mcpCall("mcp__zeta__x", nil)
fx["calls"] = results
fx["prompt_line_live"] = mcpPromptLine(mcpTools())
got, err := parityJSON(fx)
if err != nil {
t.Fatal(err)
}
got = []byte(strings.ReplaceAll(string(got), ts.URL, "{{URL}}"))
if os.Getenv("LOKI_PARITY_UPDATE") == "1" {
if err := os.MkdirAll(filepath.Dir(parityMCPFile), 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(parityMCPFile, got, 0o644); err != nil {
t.Fatal(err)
}
return
}
want, err := os.ReadFile(parityMCPFile)
if err != nil {
t.Fatalf("journal de référence absent (%v) : LOKI_PARITY_UPDATE=1 pour le créer", err)
}
if string(got) != string(want) {
t.Errorf("le client MCP ne produit plus le journal rangé dans %s.\n"+
"Si c'est voulu : LOKI_PARITY_UPDATE=1 go test ./internal/loki -run TestPariteMCP,\n"+
"puis cargo test --test mcp dit si le port Rust suit.\n--- obtenu ---\n%s", parityMCPFile, got)
}
}
func errString(err error) string {
if err == nil {
return ""
}
return err.Error()
}
+3
View File
@@ -48,6 +48,8 @@ pub struct Ctx {
pub system: Option<std::sync::Arc<crate::system::System>>,
/// Hub des projets tenu par la façade (`LOKI_PROJECTS=rust`).
pub projects: bool,
/// Serveurs MCP tenus par la façade (`LOKI_MCP=rust`).
pub mcp: Option<std::sync::Arc<crate::mcp::Mcp>>,
/// Jeton de l'API interne : une requête qui le porte vient du Go.
pub internal_token: String,
}
@@ -67,6 +69,7 @@ impl Ctx {
|| self.engine_ctl.is_some()
|| self.system.is_some()
|| self.projects
|| self.mcp.is_some()
}
/// La requête vient du Go (jeton interne valide) : déjà authentifiée par
+1
View File
@@ -25,6 +25,7 @@ pub mod engstatus;
pub mod gguf;
pub mod gojson;
pub mod hf;
pub mod mcp;
pub mod memory;
pub mod models;
pub mod oai;
+7 -1
View File
@@ -156,6 +156,9 @@ fn serve() -> ExitCode {
let system_rust = held_by_facade("LOKI_SYSTEM");
// LOKI_PROJECTS=rust : le hub des projets.
let projects_rust = held_by_facade("LOKI_PROJECTS");
// LOKI_MCP=rust : les serveurs MCP (le Go demande alors à la façade
// les outils MCP et leurs appels).
let mcp_rust = held_by_facade("LOKI_MCP");
let engine_rust = held_by_facade("LOKI_ENGINE")
&& match loki_rs::enginectl::usable(
std::env::var("LOKI_CONTAINER").ok().as_deref(),
@@ -185,6 +188,7 @@ fn serve() -> ExitCode {
("LOKI_ENGINE", engine_rust),
("LOKI_SYSTEM", system_rust),
("LOKI_PROJECTS", projects_rust),
("LOKI_MCP", mcp_rust),
];
for (var, on) in held {
if on {
@@ -284,11 +288,13 @@ fn serve() -> ExitCode {
settings: settings_rust,
files,
push: push_rust.then(|| std::sync::Arc::new(loki_rs::push::Push::new())),
mem: mem_rust.then(|| std::sync::Arc::new(loki_rs::memory::Memory::new(mem_home))),
mem: mem_rust
.then(|| std::sync::Arc::new(loki_rs::memory::Memory::new(mem_home.clone()))),
presets: presets_rust,
engine_ctl,
system: system_rust.then(|| std::sync::Arc::new(loki_rs::system::System::new())),
projects: projects_rust,
mcp: mcp_rust.then(|| std::sync::Arc::new(loki_rs::mcp::Mcp::new(mem_home))),
};
let res = proxy::serve(listener, cfg, shutdown_signal()).await;
eprintln!("[loki-rs] arrêt demandé — arrêt du serveur Go");
+440
View File
@@ -0,0 +1,440 @@
//! Client MCP (Model Context Protocol) : JSON-RPC 2.0 sur deux transports,
//! comme le SDK Go qu'utilisait `mcp_client.go` :
//!
//! - stdio : un processus local, un message JSON par ligne sur stdin/stdout ;
//! - Streamable HTTP : un POST par message, réponse en JSON ou en flux SSE,
//! session suivie par l'en-tête `Mcp-Session-Id`.
//!
//! La session s'ouvre par `initialize` puis `notifications/initialized` ; le
//! client ne sert que `tools/list` (première page, comme le Go) et
//! `tools/call`. Une requête du serveur (`ping`) reçoit une réponse vide, les
//! autres « méthode inconnue ».
use std::collections::{BTreeMap, HashMap};
use std::sync::atomic::{AtomicBool, AtomicI64, Ordering};
use std::sync::{Arc, Mutex};
use serde_json::{Map, Value, json};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, ChildStdin};
use tokio::sync::oneshot;
use super::config::ServerConfig;
/// Version du protocole demandée (celle du SDK Go 1.6).
pub const PROTOCOL_VERSION: &str = "2025-11-25";
type Pending = Arc<Mutex<HashMap<i64, oneshot::Sender<Value>>>>;
struct Stdio {
stdin: Arc<tokio::sync::Mutex<ChildStdin>>,
pending: Pending,
child: tokio::sync::Mutex<Option<Child>>,
closed: Arc<AtomicBool>,
}
struct Http {
client: reqwest::Client,
url: String,
headers: BTreeMap<String, String>,
session_id: Mutex<Option<String>>,
protocol: Mutex<Option<String>>,
}
enum Transport {
Stdio(Stdio),
Http(Http),
}
/// Une session MCP ouverte.
pub struct Session {
transport: Transport,
next_id: AtomicI64,
}
/// Le message d'erreur d'`exec` du Go pour un binaire introuvable.
fn spawn_error(cmd: &str, e: &std::io::Error) -> String {
if e.kind() == std::io::ErrorKind::NotFound {
if cmd.contains('/') {
return format!("fork/exec {cmd}: no such file or directory");
}
return format!("exec: \"{cmd}\": executable file not found in $PATH");
}
e.to_string()
}
/// Message d'une erreur JSON-RPC (`{code, message}`).
fn rpc_error(method: &str, err: &Value) -> String {
let msg = err
.get("message")
.and_then(Value::as_str)
.unwrap_or("erreur JSON-RPC");
format!("calling \"{method}\": {msg}")
}
impl Session {
/// Ouvre la session : processus ou point d'accès HTTP, puis la poignée de
/// main. `version` : celle annoncée dans `clientInfo`.
pub async fn connect(cfg: &ServerConfig, version: &str) -> Result<Session, String> {
let transport = match cfg.transport() {
"stdio" => Transport::Stdio(spawn(cfg)?),
"http" => {
crate::chat::engine::install_crypto();
Transport::Http(Http {
client: reqwest::Client::builder()
.build()
.map_err(|e| e.to_string())?,
url: cfg.url.trim().to_string(),
headers: cfg.headers.clone(),
session_id: Mutex::new(None),
protocol: Mutex::new(None),
})
}
_ => return Err("serveur MCP mal configuré (ni command ni url)".into()),
};
let s = Session {
transport,
next_id: AtomicI64::new(1),
};
let init = s
.request(
"initialize",
json!({
"protocolVersion": PROTOCOL_VERSION,
"capabilities": {},
"clientInfo": { "name": "loki", "version": version },
}),
)
.await;
let init = match init {
Ok(v) => v,
Err(e) => {
s.close().await;
return Err(e);
}
};
if let Transport::Http(h) = &s.transport {
let v = init
.get("protocolVersion")
.and_then(Value::as_str)
.unwrap_or(PROTOCOL_VERSION)
.to_string();
*h.protocol.lock().unwrap_or_else(|e| e.into_inner()) = Some(v);
}
if let Err(e) = s.notify("notifications/initialized", json!({})).await {
s.close().await;
return Err(e);
}
Ok(s)
}
/// `tools/list` (première page, comme le client Go).
pub async fn list_tools(&self) -> Result<Vec<Value>, String> {
let r = self.request("tools/list", json!({})).await?;
Ok(r.get("tools")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default())
}
/// `tools/call` : le résultat brut (`content`, `structuredContent`,
/// `isError`).
pub async fn call_tool(&self, name: &str, args: &Map<String, Value>) -> Result<Value, String> {
self.request(
"tools/call",
json!({ "name": name, "arguments": Value::Object(args.clone()) }),
)
.await
}
/// La session est-elle tombée (processus mort) ?
pub fn is_closed(&self) -> bool {
match &self.transport {
Transport::Stdio(s) => s.closed.load(Ordering::SeqCst),
Transport::Http(_) => false,
}
}
/// Ferme la session : processus tué, ou session HTTP terminée.
pub async fn close(&self) {
match &self.transport {
Transport::Stdio(s) => {
s.closed.store(true, Ordering::SeqCst);
if let Some(mut c) = s.child.lock().await.take() {
let _ = c.kill().await;
}
}
Transport::Http(h) => {
let sid = h
.session_id
.lock()
.unwrap_or_else(|e| e.into_inner())
.clone();
if let Some(sid) = sid {
let mut rq = h.client.delete(&h.url).header("Mcp-Session-Id", sid);
for (k, v) in &h.headers {
rq = rq.header(k.as_str(), v.as_str());
}
let _ =
tokio::time::timeout(std::time::Duration::from_secs(5), rq.send()).await;
}
}
}
}
async fn request(&self, method: &str, params: Value) -> Result<Value, String> {
let id = self.next_id.fetch_add(1, Ordering::SeqCst);
let msg = json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params });
let resp = match &self.transport {
Transport::Stdio(s) => s.request(id, method, &msg).await?,
Transport::Http(h) => h.request(id, method, &msg).await?,
};
if let Some(err) = resp.get("error").filter(|e| !e.is_null()) {
return Err(rpc_error(method, err));
}
Ok(resp.get("result").cloned().unwrap_or(Value::Null))
}
async fn notify(&self, method: &str, params: Value) -> Result<(), String> {
let msg = json!({ "jsonrpc": "2.0", "method": method, "params": params });
match &self.transport {
Transport::Stdio(s) => s.write(&msg).await,
Transport::Http(h) => h.post(&msg).await.map(|_| ()),
}
}
}
impl Drop for Session {
fn drop(&mut self) {
if let Transport::Stdio(s) = &self.transport {
s.closed.store(true, Ordering::SeqCst);
if let Ok(mut g) = s.child.try_lock()
&& let Some(c) = g.as_mut()
{
let _ = c.start_kill();
}
}
}
}
// ---------------------------------------------------------------------------
// stdio
// ---------------------------------------------------------------------------
fn spawn(cfg: &ServerConfig) -> Result<Stdio, String> {
let mut cmd = tokio::process::Command::new(&cfg.command);
cmd.args(&cfg.args)
.envs(&cfg.env)
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::null())
.kill_on_drop(true);
let mut child = cmd.spawn().map_err(|e| spawn_error(&cfg.command, &e))?;
let stdin = Arc::new(tokio::sync::Mutex::new(
child.stdin.take().ok_or("stdin indisponible")?,
));
let stdout = child.stdout.take().ok_or("stdout indisponible")?;
let pending: Pending = Arc::new(Mutex::new(HashMap::new()));
let closed = Arc::new(AtomicBool::new(false));
let (p2, c2, w2) = (pending.clone(), closed.clone(), stdin.clone());
tokio::spawn(async move {
let mut lines = BufReader::new(stdout).lines();
while let Ok(Some(line)) = lines.next_line().await {
let Ok(v) = serde_json::from_str::<Value>(&line) else {
continue;
};
if let Some(reply) = dispatch(v, &p2) {
let mut b = serde_json::to_vec(&reply).unwrap_or_default();
b.push(b'\n');
let mut w = w2.lock().await;
let _ = w.write_all(&b).await;
let _ = w.flush().await;
}
}
c2.store(true, Ordering::SeqCst);
// Les appels en attente échouent : leur émetteur disparaît.
p2.lock().unwrap_or_else(|e| e.into_inner()).clear();
});
Ok(Stdio {
stdin,
pending,
child: tokio::sync::Mutex::new(Some(child)),
closed,
})
}
/// Une ligne lue : réponse à un appel en attente (rien à renvoyer), requête
/// du serveur (la réponse à lui écrire), ou notification (ignorée).
fn dispatch(v: Value, pending: &Pending) -> Option<Value> {
let has_method = v.get("method").is_some();
match (v.get("id"), has_method) {
(Some(id), false) => {
if let Some(n) = id.as_i64()
&& let Some(tx) = pending.lock().unwrap_or_else(|e| e.into_inner()).remove(&n)
{
let _ = tx.send(v);
}
None
}
(Some(id), true) => Some(if v.get("method").and_then(Value::as_str) == Some("ping") {
json!({ "jsonrpc": "2.0", "id": id, "result": {} })
} else {
json!({ "jsonrpc": "2.0", "id": id,
"error": { "code": -32601, "message": "method not found" } })
}),
_ => None,
}
}
impl Stdio {
async fn write(&self, msg: &Value) -> Result<(), String> {
if self.closed.load(Ordering::SeqCst) {
return Err("connection closed".into());
}
let mut line = serde_json::to_vec(msg).map_err(|e| e.to_string())?;
line.push(b'\n');
let mut w = self.stdin.lock().await;
w.write_all(&line).await.map_err(|e| e.to_string())?;
w.flush().await.map_err(|e| e.to_string())
}
async fn request(&self, id: i64, method: &str, msg: &Value) -> Result<Value, String> {
let (tx, rx) = oneshot::channel();
self.pending
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(id, tx);
if let Err(e) = self.write(msg).await {
self.pending
.lock()
.unwrap_or_else(|e| e.into_inner())
.remove(&id);
return Err(format!("calling \"{method}\": {e}"));
}
rx.await
.map_err(|_| format!("calling \"{method}\": connection closed"))
}
}
// ---------------------------------------------------------------------------
// Streamable HTTP
// ---------------------------------------------------------------------------
impl Http {
fn builder(&self, msg: &Value) -> reqwest::RequestBuilder {
let mut rq = self
.client
.post(&self.url)
.header("Content-Type", "application/json")
.header("Accept", "application/json, text/event-stream")
.body(serde_json::to_vec(msg).unwrap_or_default());
if let Some(sid) = self
.session_id
.lock()
.unwrap_or_else(|e| e.into_inner())
.clone()
{
rq = rq.header("Mcp-Session-Id", sid);
}
if let Some(p) = self
.protocol
.lock()
.unwrap_or_else(|e| e.into_inner())
.clone()
{
rq = rq.header("MCP-Protocol-Version", p);
}
for (k, v) in &self.headers {
rq = rq.header(k.as_str(), v.as_str());
}
rq
}
async fn post(&self, msg: &Value) -> Result<reqwest::Response, String> {
let resp = self.builder(msg).send().await.map_err(|e| e.to_string())?;
if let Some(sid) = resp
.headers()
.get("mcp-session-id")
.and_then(|v| v.to_str().ok())
{
*self.session_id.lock().unwrap_or_else(|e| e.into_inner()) = Some(sid.to_string());
}
if !resp.status().is_success() {
let st = resp.status();
let body = resp.text().await.unwrap_or_default();
let body = body.trim();
return Err(if body.is_empty() {
st.to_string()
} else {
format!("{st}: {body}")
});
}
Ok(resp)
}
async fn request(&self, id: i64, method: &str, msg: &Value) -> Result<Value, String> {
let resp = self
.post(msg)
.await
.map_err(|e| format!("calling \"{method}\": {e}"))?;
let ctype = resp
.headers()
.get("content-type")
.and_then(|v| v.to_str().ok())
.unwrap_or("")
.to_ascii_lowercase();
let body = resp
.text()
.await
.map_err(|e| format!("calling \"{method}\": {e}"))?;
let mut candidates: Vec<Value> = Vec::new();
if ctype.starts_with("text/event-stream") {
let mut data = String::new();
for line in body.lines().chain(std::iter::once("")) {
if line.is_empty() {
if !data.is_empty() {
if let Ok(v) = serde_json::from_str::<Value>(&data) {
candidates.push(v);
}
data.clear();
}
} else if let Some(d) = line.strip_prefix("data:") {
if !data.is_empty() {
data.push('\n');
}
data.push_str(d.strip_prefix(' ').unwrap_or(d));
}
}
} else if let Ok(v) = serde_json::from_str::<Value>(&body) {
match v {
Value::Array(a) => candidates.extend(a),
v => candidates.push(v),
}
}
candidates
.into_iter()
.find(|v| v.get("id").and_then(Value::as_i64) == Some(id) && v.get("method").is_none())
.ok_or_else(|| format!("calling \"{method}\": réponse absente"))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn erreurs_comme_go() {
let e = std::io::Error::from(std::io::ErrorKind::NotFound);
assert_eq!(
spawn_error("bidon", &e),
"exec: \"bidon\": executable file not found in $PATH"
);
assert_eq!(
rpc_error(
"tools/call",
&json!({"code": -32602, "message": "outil inconnu"})
),
"calling \"tools/call\": outil inconnu"
);
}
}
+483
View File
@@ -0,0 +1,483 @@
//! Configuration des serveurs MCP (`mcp_config.go`) et fonctions pures du
//! client (`mcp_client.go`) : nommage des outils, schémas, libellés, ligne du
//! préambule, délais.
use std::collections::BTreeMap;
use std::time::Duration;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value, json};
/// `MCPServerConfig` : un seul transport renseigné, `command` (stdio) ou
/// `url` (Streamable HTTP).
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ServerConfig {
#[serde(skip_serializing_if = "String::is_empty")]
pub command: String,
#[serde(skip_serializing_if = "Vec::is_empty")]
pub args: Vec<String>,
#[serde(skip_serializing_if = "BTreeMap::is_empty")]
pub env: BTreeMap<String, String>,
#[serde(skip_serializing_if = "String::is_empty")]
pub url: String,
#[serde(skip_serializing_if = "BTreeMap::is_empty")]
pub headers: BTreeMap<String, String>,
pub enabled: bool,
#[serde(rename = "disabledTools", skip_serializing_if = "Vec::is_empty")]
pub disabled_tools: Vec<String>,
}
impl Default for ServerConfig {
fn default() -> Self {
ServerConfig {
command: String::new(),
args: Vec::new(),
env: BTreeMap::new(),
url: String::new(),
headers: BTreeMap::new(),
enabled: true,
disabled_tools: Vec::new(),
}
}
}
/// Lecture tolérante : `enabled` vaut vrai quand la clé est absente (fichiers
/// à la Claude Desktop), les champs mal typés restent vides.
impl<'de> Deserialize<'de> for ServerConfig {
fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
let v = Map::<String, Value>::deserialize(d)?;
let s = |k: &str| v.get(k).and_then(Value::as_str).unwrap_or("").to_string();
let list = |k: &str| -> Vec<String> {
v.get(k)
.and_then(Value::as_array)
.map(|a| {
a.iter()
.filter_map(|x| x.as_str().map(str::to_string))
.collect()
})
.unwrap_or_default()
};
let map = |k: &str| -> BTreeMap<String, String> {
v.get(k)
.and_then(Value::as_object)
.map(|o| {
o.iter()
.filter_map(|(k, x)| x.as_str().map(|s| (k.clone(), s.to_string())))
.collect()
})
.unwrap_or_default()
};
Ok(ServerConfig {
command: s("command"),
args: list("args"),
env: map("env"),
url: s("url"),
headers: map("headers"),
enabled: match v.get("enabled") {
None => true,
Some(x) => x.as_bool().unwrap_or(false),
},
disabled_tools: list("disabledTools"),
})
}
}
impl ServerConfig {
/// `Transport` : « stdio », « http » ou « » (mal configuré).
pub fn transport(&self) -> &'static str {
if !self.command.trim().is_empty() {
"stdio"
} else if !self.url.trim().is_empty() {
"http"
} else {
""
}
}
/// `Validate` : exactement un transport.
pub fn validate(&self) -> Result<(), String> {
let (cmd, url) = (!self.command.trim().is_empty(), !self.url.trim().is_empty());
match (cmd, url) {
(true, true) => Err(
"un serveur MCP ne peut avoir à la fois 'command' (stdio) et 'url' (http)".into(),
),
(false, false) => {
Err("un serveur MCP doit avoir soit 'command' (stdio) soit 'url' (http)".into())
}
_ => Ok(()),
}
}
/// `ToolDisabled`.
pub fn tool_disabled(&self, tool: &str) -> bool {
self.disabled_tools.iter().any(|t| t == tool)
}
}
/// Les serveurs déclarés (`state/mcp`), triés par nom.
pub type Servers = BTreeMap<String, ServerConfig>;
/// `loadMCPConfigLocked` : illisible ou absent → aucun serveur.
pub fn decode_servers(raw: Option<&[u8]>) -> Servers {
raw.and_then(|b| serde_json::from_slice::<Option<Servers>>(b).ok())
.flatten()
.unwrap_or_default()
}
/// `SetMCPServer` (sans l'invalidation du pool) : nom contrôlé, transport
/// validé.
pub fn set_server(servers: &mut Servers, name: &str, cfg: ServerConfig) -> Result<(), String> {
let name = name.trim();
if name.is_empty() {
return Err("nom de serveur vide".into());
}
if name.contains("__") {
return Err("le nom ne peut pas contenir '__' (réservé au namespacing des outils)".into());
}
cfg.validate()?;
servers.insert(name.to_string(), cfg);
Ok(())
}
fn unknown(name: &str) -> String {
format!("serveur MCP inconnu: {name}")
}
/// `DeleteMCPServer`.
pub fn delete_server(servers: &mut Servers, name: &str) -> Result<(), String> {
servers
.remove(name)
.map(|_| ())
.ok_or_else(|| unknown(name))
}
/// `SetMCPServerEnabled`.
pub fn set_enabled(servers: &mut Servers, name: &str, on: bool) -> Result<(), String> {
let cfg = servers.get_mut(name).ok_or_else(|| unknown(name))?;
cfg.enabled = on;
Ok(())
}
/// `SetMCPToolEnabled` : la liste reconstruite sans l'outil, puis l'outil
/// ajouté en fin s'il est masqué.
pub fn set_tool_enabled(
servers: &mut Servers,
server: &str,
tool: &str,
on: bool,
) -> Result<(), String> {
let cfg = servers.get_mut(server).ok_or_else(|| unknown(server))?;
cfg.disabled_tools.retain(|t| t != tool);
if !on {
cfg.disabled_tools.push(tool.to_string());
}
Ok(())
}
// ---------------------------------------------------------------------------
// Fonctions pures du client
// ---------------------------------------------------------------------------
/// `mcpSanitize` : `[^a-zA-Z0-9_-]+` → `_`, puis `_` retirés aux bords.
pub fn sanitize(s: &str) -> String {
let mut out = String::with_capacity(s.len());
let mut in_run = false;
for c in s.chars() {
if c.is_ascii_alphanumeric() || c == '_' || c == '-' {
out.push(c);
in_run = false;
} else if !in_run {
out.push('_');
in_run = true;
}
}
out.trim_matches('_').to_string()
}
/// `mcpExposedName` : `mcp__<serveur>__<outil>`.
pub fn exposed_name(server: &str, tool: &str) -> String {
format!("mcp__{}__{}", sanitize(server), sanitize(tool))
}
/// `isMCPTool`.
pub fn is_mcp_tool(name: &str) -> bool {
name.starts_with("mcp__")
}
/// `mcpNormalizeSchema` : un objet tel quel, sinon un objet sans propriété.
pub fn normalize_schema(schema: Option<&Value>) -> Value {
match schema {
Some(Value::Object(m)) => Value::Object(m.clone()),
_ => json!({ "type": "object", "properties": {} }),
}
}
/// `mcpArgLabel` : une clé parlante, sinon le JSON des arguments (120 runes).
pub fn arg_label(args: &Map<String, Value>) -> String {
if args.is_empty() {
return String::new();
}
for k in ["path", "file", "query", "command", "url", "name"] {
if let Some(Value::String(v)) = args.get(k)
&& !v.is_empty()
{
return v.clone();
}
}
let s = crate::gojson::marshal(args);
if s.chars().count() > 120 {
return s.chars().take(120).collect::<String>() + "…";
}
s
}
/// `mcpTagDescription`.
pub fn tag_description(server: &str, desc: &str) -> String {
format!("[MCP: {server}] {desc}")
}
/// `mcpToolServer` : le serveur d'origine d'un outil annoncé, d'après le
/// préfixe de sa description (la première coupure dont la forme assainie
/// figure bien dans le nom de l'outil).
pub fn tool_server(name: &str, description: &str) -> Option<String> {
if !name.starts_with("mcp__") {
return None;
}
let rest = description.strip_prefix("[MCP: ")?;
let mut i = 0;
loop {
let j = rest[i..].find("] ")?;
let server = &rest[..i + j];
if name.starts_with(&format!("mcp__{}__", sanitize(server))) {
return Some(server.to_string());
}
i += j + 1;
}
}
/// `mcpPromptLine` : les serveurs des outils annoncés ce tour-ci, et leur
/// nombre d'outils. `tools` : (nom, description).
pub fn prompt_line(tools: &[(String, String)]) -> String {
let mut counts: BTreeMap<String, usize> = BTreeMap::new();
for (n, d) in tools {
if let Some(s) = tool_server(n, d) {
*counts.entry(s).or_default() += 1;
}
}
if counts.is_empty() {
return String::new();
}
let mut names: Vec<String> = counts.iter().map(|(s, n)| format!("{s} ({n})")).collect();
names.sort();
format!(
"\n\nMCP servers connected (their tools are prefixed mcp__<server>__<tool>): {}.",
names.join(", ")
)
}
/// `mcpConnectTimeout`.
pub const CONNECT_TIMEOUT: Duration = Duration::from_secs(20);
/// `mcpConnectTimeoutCold` : npx et uvx téléchargent le paquet au premier
/// lancement.
pub const CONNECT_TIMEOUT_COLD: Duration = Duration::from_secs(180);
/// `mcpCallTimeout`.
pub const CALL_TIMEOUT: Duration = Duration::from_secs(120);
/// Nom du lanceur (`filepath.Base`), en minuscules, sans `.exe` puis sans
/// `.cmd`.
fn launcher(cfg: &ServerConfig) -> String {
let t = cfg.command.trim_end_matches('/');
let base = if t.is_empty() {
if cfg.command.is_empty() { "." } else { "/" }
} else {
t.rsplit('/').next().unwrap_or(t)
};
let base = base.to_lowercase();
let b = base.strip_suffix(".exe").unwrap_or(&base);
b.strip_suffix(".cmd").unwrap_or(b).to_string()
}
/// `mcpTimeoutFor`.
pub fn timeout_for(cfg: &ServerConfig) -> Duration {
match launcher(cfg).as_str() {
"npx" | "uvx" => CONNECT_TIMEOUT_COLD,
_ => CONNECT_TIMEOUT,
}
}
/// `mcpFriendlyErr` pour un délai dépassé.
pub fn timeout_error(cfg: &ServerConfig) -> String {
let l = launcher(cfg);
if l == "npx" || l == "uvx" {
return format!(
"délai dépassé — {l} télécharge le paquet du serveur au premier lancement, ce qui peut être long ; clique « tester » pour réessayer (le paquet reste en cache), et vérifie que le conteneur atteint le registre (npm/PyPI)"
);
}
"délai dépassé — le serveur n'a pas répondu au handshake en 20s".into()
}
/// `mcpOutputCap` : environ un caractère par jeton de contexte, jamais sous
/// 12 000.
pub fn output_cap(ctx: i64) -> usize {
12000.max(ctx.max(0) as usize)
}
/// `ctxWindow` : `CTX` de la configuration, 32 768 à défaut.
pub fn ctx_window(cfg: &std::collections::HashMap<String, String>) -> i64 {
cfg.get("CTX")
.and_then(|v| v.parse::<i64>().ok())
.filter(|n| *n > 0)
.unwrap_or(32768)
}
/// Le texte d'un résultat d'outil, tronqué au plafond (en runes).
pub fn cap_output(out: String, lim: usize) -> String {
let n = out.chars().count();
if n <= lim {
return out;
}
let kept: String = out.chars().take(lim).collect();
format!("{kept}\n…[tronqué : {n} caractères au total, {lim} gardés]")
}
/// `flattenMCPContent` : blocs texte concaténés, images et sons résumés, les
/// autres rendus en JSON ; le contenu structuré en JSON indenté s'il n'y a pas
/// de texte.
pub fn flatten_content(result: &Value) -> String {
let mut parts: Vec<String> = Vec::new();
for c in result
.get("content")
.and_then(Value::as_array)
.into_iter()
.flatten()
{
let kind = c.get("type").and_then(Value::as_str).unwrap_or("");
let mime = c.get("mimeType").and_then(Value::as_str).unwrap_or("");
match kind {
"text" => parts.push(c.get("text").and_then(Value::as_str).unwrap_or("").into()),
"image" => parts.push(format!("[image {mime}]")),
"audio" => parts.push(format!("[audio {mime}]")),
_ => parts.push(wire_content_json(c)),
}
}
let text = parts.join("\n").trim().to_string();
if text.is_empty()
&& let Some(sc) = result.get("structuredContent").filter(|v| !v.is_null())
{
return String::from_utf8(crate::chatx::marshal_indent(sc)).unwrap_or_default();
}
text
}
/// `json.Marshal` d'un contenu « autre » par le SDK Go (`wireContent`) :
/// champs dans l'ordre de sa struct, vides omis.
fn wire_content_json(c: &Value) -> String {
let order = [
"type",
"text",
"mimeType",
"data",
"resource",
"uri",
"name",
"title",
"description",
"size",
"_meta",
"annotations",
"icons",
"id",
"input",
"toolUseId",
"content",
"structuredContent",
"isError",
];
let mut out = String::from("{");
let mut first = true;
for k in order {
let Some(v) = c.get(k) else { continue };
let empty = match v {
Value::Null => true,
Value::String(s) => s.is_empty(),
Value::Bool(b) => !b,
Value::Array(a) => a.is_empty(),
Value::Object(o) => o.is_empty() && k != "resource",
Value::Number(_) => false,
};
if empty && k != "type" {
continue;
}
if !first {
out.push(',');
}
first = false;
out.push_str(&crate::gojson::marshal(k));
out.push(':');
if k == "resource" {
out.push_str(&resource_contents_json(v));
} else {
out.push_str(&crate::gojson::marshal(v));
}
}
out.push('}');
out
}
/// `ResourceContents` du SDK Go : `uri` toujours, puis `mimeType`, `text`,
/// `blob`, `_meta` s'ils sont renseignés.
fn resource_contents_json(v: &Value) -> String {
let mut out = String::from("{");
let mut first = true;
for (k, always) in [
("uri", true),
("mimeType", false),
("text", false),
("blob", false),
("_meta", false),
] {
let x = v.get(k).cloned().unwrap_or(Value::Null);
let empty = match &x {
Value::Null => true,
Value::String(s) => s.is_empty(),
Value::Object(o) => o.is_empty(),
_ => false,
};
if empty && !always {
continue;
}
if !first {
out.push(',');
}
first = false;
out.push_str(&crate::gojson::marshal(k));
out.push(':');
let x = if x.is_null() && always { json!("") } else { x };
out.push_str(&crate::gojson::marshal(&x));
}
out.push('}');
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn noms() {
assert_eq!(sanitize("mon serveur!"), "mon_serveur");
assert_eq!(exposed_name("a b", "c.d"), "mcp__a_b__c_d");
}
#[test]
fn lanceurs() {
let c = |cmd: &str| ServerConfig {
command: cmd.into(),
..Default::default()
};
assert_eq!(timeout_for(&c("/usr/bin/npx")), CONNECT_TIMEOUT_COLD);
assert_eq!(timeout_for(&c("UVX.EXE")), CONNECT_TIMEOUT_COLD);
assert_eq!(timeout_for(&c("npx.cmd")), CONNECT_TIMEOUT_COLD);
assert_eq!(timeout_for(&c("node")), CONNECT_TIMEOUT);
}
}
+684
View File
@@ -0,0 +1,684 @@
//! Serveurs MCP tenus par la façade (lot 20 de la migration,
//! `LOKI_MCP=rust` ; expérimental, le Go reste le défaut) — port de
//! `mcp_client.go`, `mcp_config.go`, `mcp_catalog.go` et des routes
//! `/api/mcp*` de `web_api.go`.
//!
//! Le pool de sessions vit ici : une session reste ouverte d'un tour à
//! l'autre (un serveur stdio garde son processus), les connexions se font en
//! parallèle, deux appelants simultanés partagent la même tentative. Le Go,
//! qui exécute encore les outils de l'agent, demande à la façade la liste des
//! outils MCP et leurs appels (`/internal/facade/v1/mcp/*`) : un seul pool,
//! quelle que soit la conversation qui tourne.
pub mod client;
pub mod config;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, LazyLock, Mutex};
use futures_util::future::join_all;
use http_body_util::{BodyExt, Limited};
use hyper::body::Incoming;
use hyper::{Request, Response, StatusCode};
use serde_json::{Map, Value, json};
use crate::api::Ctx;
use crate::auth::json as send_json;
use crate::gojson::{self, Kind};
use crate::proxy::Body;
use crate::store::{BK_STATE, Store};
use client::Session;
use config::{ServerConfig, Servers};
/// Clé de la configuration (`state/mcp`).
pub const ST_MCP: &str = "mcp";
/// Chemins de l'API interne de la façade, pour le Go.
pub const INTERNAL_TOOLS: &str = "/internal/facade/v1/mcp/tools";
pub const INTERNAL_CALL: &str = "/internal/facade/v1/mcp/call";
pub const INTERNAL_PREWARM: &str = "/internal/facade/v1/mcp/prewarm";
/// Issue d'une tentative de connexion.
#[derive(Default)]
struct SlotState {
sess: Option<Arc<Session>>,
tools: Vec<Value>,
err: Option<String>,
}
/// Une entrée du pool, publiée AVANT la connexion : un second appelant attend
/// son issue au lieu de lancer un deuxième processus.
struct Slot {
cfg: ServerConfig,
state: tokio::sync::OnceCell<SlotState>,
}
/// Le pool des sessions MCP et le registre nom exposé → (serveur, outil).
pub struct Mcp {
home: PathBuf,
slots: Mutex<HashMap<String, Arc<Slot>>>,
registry: Mutex<HashMap<String, (String, String)>>,
cfg_lock: tokio::sync::Mutex<()>,
}
fn lock<T>(m: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
m.lock().unwrap_or_else(|e| e.into_inner())
}
impl Mcp {
pub fn new(home: PathBuf) -> Self {
Mcp {
home,
slots: Mutex::new(HashMap::new()),
registry: Mutex::new(HashMap::new()),
cfg_lock: tokio::sync::Mutex::new(()),
}
}
/// `LoadMCPConfig`.
pub async fn load(store: &Store) -> Servers {
config::decode_servers(store.get(BK_STATE, ST_MCP).await.ok().flatten().as_deref())
}
async fn save(store: &Store, servers: &Servers) -> Result<(), String> {
store
.put(BK_STATE, ST_MCP, gojson::marshal(servers).into_bytes())
.await
}
/// Une mutation de la configuration, sérialisée, puis l'invalidation du
/// serveur touché.
async fn mutate(
&self,
store: &Store,
name: &str,
f: impl FnOnce(&mut Servers) -> Result<(), String>,
) -> Result<(), String> {
{
let _g = self.cfg_lock.lock().await;
let mut servers = Self::load(store).await;
f(&mut servers)?;
Self::save(store, &servers).await?;
}
self.invalidate(name).await;
Ok(())
}
/// `mcpInvalidate` : la session du serveur fermée et oubliée, ses outils
/// retirés du registre.
pub async fn invalidate(&self, name: &str) {
let slot = lock(&self.slots).remove(name);
lock(&self.registry).retain(|_, (server, _)| server != name);
// Une session encore en cours de connexion se refermera d'elle-même en
// constatant qu'elle n'est plus dans le pool.
if let Some(s) = slot
&& let Some(st) = s.state.get()
&& let Some(sess) = &st.sess
{
sess.close().await;
}
}
/// `mcpCloseAll`.
pub async fn close_all(&self) {
let slots: Vec<_> = lock(&self.slots).drain().map(|(_, s)| s).collect();
lock(&self.registry).clear();
for s in slots {
if let Some(st) = s.state.get()
&& let Some(sess) = &st.sess
{
sess.close().await;
}
}
}
/// `ensure` : la session du serveur, connectée ou en erreur.
async fn ensure(&self, name: &str, cfg: &ServerConfig, version: &str) -> Arc<Slot> {
let slot = lock(&self.slots)
.entry(name.to_string())
.or_insert_with(|| {
Arc::new(Slot {
cfg: cfg.clone(),
state: tokio::sync::OnceCell::new(),
})
})
.clone();
slot.state
.get_or_init(|| async {
let cfg = &slot.cfg;
let mut st = SlotState::default();
let work = async {
let sess = Session::connect(cfg, version).await?;
let tools = sess.list_tools().await;
Ok::<_, String>((sess, tools))
};
match tokio::time::timeout(config::timeout_for(cfg), work).await {
Err(_) => st.err = Some(config::timeout_error(cfg)),
Ok(Err(e)) => st.err = Some(e),
Ok(Ok((sess, tools))) => {
match tools {
Ok(t) => st.tools = t,
Err(e) => st.err = Some(e),
}
st.sess = Some(Arc::new(sess));
}
}
// Invalidée pendant la connexion : jamais republiée, refermée.
let current = lock(&self.slots)
.get(name)
.is_some_and(|s| Arc::ptr_eq(s, &slot));
if current {
let mut reg = lock(&self.registry);
reg.retain(|_, (server, _)| server != name);
for t in &st.tools {
let tn = t.get("name").and_then(Value::as_str).unwrap_or("");
reg.insert(
config::exposed_name(name, tn),
(name.to_string(), tn.to_string()),
);
}
} else if let Some(sess) = st.sess.take() {
sess.close().await;
}
st
})
.await;
slot
}
/// `mcpEnsureAll` : les serveurs activés connectés en parallèle.
async fn ensure_all(&self, servers: &Servers, version: &str) {
join_all(
servers
.iter()
.filter(|(_, c)| c.enabled && !c.transport().is_empty())
.map(|(n, c)| self.ensure(n, c, version)),
)
.await;
}
/// `mcpTools` : les outils des serveurs activés, prêts à être annoncés au
/// modèle (forme `Tool` du Go).
pub async fn tools(&self, store: &Store, version: &str) -> Vec<Value> {
let servers = Self::load(store).await;
if servers.is_empty() {
return Vec::new();
}
self.ensure_all(&servers, version).await;
let mut out = Vec::new();
for (name, cfg) in &servers {
if !cfg.enabled || cfg.transport().is_empty() {
continue;
}
let slot = self.ensure(name, cfg, version).await;
let Some(st) = slot.state.get() else { continue };
if st.sess.is_none() {
continue;
}
for t in &st.tools {
let tn = t.get("name").and_then(Value::as_str).unwrap_or("");
if cfg.tool_disabled(tn) {
continue;
}
let mut desc = t.get("description").and_then(Value::as_str).unwrap_or("");
if desc.is_empty() {
desc = t.get("title").and_then(Value::as_str).unwrap_or("");
}
out.push(json!({
"type": "function",
"function": {
"name": config::exposed_name(name, tn),
"description": config::tag_description(name, desc),
"parameters": config::normalize_schema(t.get("inputSchema")),
},
}));
}
}
out
}
/// `mcpCall` : un outil par son nom exposé ; une reconnexion si la
/// session est morte.
pub async fn call(
&self,
store: &Store,
name: &str,
args: &Map<String, Value>,
version: &str,
) -> String {
let Some((server, tool)) = lock(&self.registry).get(name).cloned() else {
return format!("[erreur] outil MCP inconnu ou serveur non connecté: {name}");
};
let cap = config::output_cap(config::ctx_window(&store.config().await));
let mut err = match self.call_once(&server, &tool, args, cap).await {
Ok(r) => return r,
Err(e) => e,
};
let servers = Self::load(store).await;
if let Some(cfg) = servers.get(&server).filter(|c| c.enabled) {
self.invalidate(&server).await;
self.ensure(&server, cfg, version).await;
match self.call_once(&server, &tool, args, cap).await {
Ok(r) => return r,
Err(e) => err = e,
}
}
format!("[erreur MCP] {err}")
}
async fn call_once(
&self,
server: &str,
tool: &str,
args: &Map<String, Value>,
cap: usize,
) -> Result<String, String> {
let slot = lock(&self.slots).get(server).cloned();
let sess = slot
.as_ref()
.and_then(|s| s.state.get())
.and_then(|st| st.sess.clone())
.filter(|s| !s.is_closed())
.ok_or_else(|| format!("serveur {server} non connecté"))?;
let res = tokio::time::timeout(config::CALL_TIMEOUT, sess.call_tool(tool, args))
.await
.map_err(|_| "context deadline exceeded".to_string())??;
let out = config::cap_output(config::flatten_content(&res), cap);
if res.get("isError").and_then(Value::as_bool) == Some(true) {
return Ok(format!("[l'outil a renvoyé une erreur]\n{out}"));
}
if out.trim().is_empty() {
return Ok("[ok] (aucune sortie)".into());
}
Ok(out)
}
/// `MCPStatus` : l'état de chaque serveur, connexion tentée pour les
/// activés.
pub async fn status(&self, store: &Store, version: &str) -> Vec<Value> {
let servers = Self::load(store).await;
self.ensure_all(&servers, version).await;
let mut out = Vec::new();
for (name, cfg) in &servers {
let mut st = json!({
"name": name,
"transport": cfg.transport(),
"enabled": cfg.enabled,
"connected": false,
"tools": [],
"disabled": cfg.disabled_tools,
});
for (k, v) in [
("command", json!(cfg.command)),
("args", json!(cfg.args)),
("env", json!(cfg.env)),
("url", json!(cfg.url)),
("headers", json!(cfg.headers)),
] {
let empty = match &v {
Value::String(s) => s.is_empty(),
Value::Array(a) => a.is_empty(),
Value::Object(o) => o.is_empty(),
_ => false,
};
if !empty {
st[k] = v;
}
}
if cfg.enabled && !cfg.transport().is_empty() {
let slot = self.ensure(name, cfg, version).await;
if let Some(s) = slot.state.get() {
if s.sess.is_some() {
st["connected"] = json!(true);
st["tools"] = json!(
s.tools
.iter()
.map(|t| t.get("name").and_then(Value::as_str).unwrap_or(""))
.collect::<Vec<_>>()
);
}
if let Some(e) = &s.err {
st["error"] = json!(e);
}
}
}
out.push(st);
}
out
}
/// `MCPPrewarm` : les sessions ouvertes en tâche de fond, en mode agent
/// seulement (sans lui, aucun outil MCP n'est annoncé).
pub fn prewarm(self: &Arc<Self>, store: Store, version: String) {
let me = self.clone();
tokio::spawn(async move {
if store.get_str(BK_STATE, "agent").await != "1" {
return;
}
let servers = Self::load(&store).await;
if !servers.is_empty() {
me.ensure_all(&servers, &version).await;
}
});
}
/// `MCPCatalog` : le catalogue embarqué, chemins résolus, entrées déjà
/// configurées marquées.
pub async fn catalog(&self, store: &Store) -> Result<Vec<Value>, String> {
let entries = catalog_entries()?;
let configured = Self::load(store).await;
let ws = workspace_dir(&self.home);
let expand = |s: &str| s.replace("{workspace}", &ws);
let mut out = Vec::new();
for e in entries {
let mut e = e.clone();
let obj = e.as_object_mut().ok_or("entrée du catalogue illisible")?;
if let Some(Value::Array(a)) = obj.get_mut("args") {
for x in a.iter_mut() {
if let Value::String(s) = x {
*s = expand(s);
}
}
}
if let Some(Value::Object(m)) = obj.get_mut("env") {
for x in m.values_mut() {
if let Value::String(s) = x {
*s = expand(s);
}
}
}
for k in ["homepage", "runtime"] {
if obj.get(k).and_then(Value::as_str) == Some("") {
obj.remove(k);
}
}
for k in ["args", "secrets"] {
if obj
.get(k)
.and_then(Value::as_array)
.is_some_and(|a| a.is_empty())
{
obj.remove(k);
}
}
if obj
.get("env")
.and_then(Value::as_object)
.is_some_and(|m| m.is_empty())
{
obj.remove("env");
}
let id = obj.get("id").and_then(Value::as_str).unwrap_or("");
let runtime = obj.get("runtime").and_then(Value::as_str).unwrap_or("");
let (installed, runtime_ok) = (configured.contains_key(id), runtime_available(runtime));
obj.insert("installed".into(), json!(installed));
obj.insert("runtimeOk".into(), json!(runtime_ok));
out.push(e);
}
Ok(out)
}
}
/// `workspaceDir` : le dossier de travail de l'agent.
fn workspace_dir(home: &Path) -> String {
crate::paths::go_join(&[&home.to_string_lossy(), "workspace"])
}
/// `mcpRuntimeAvailable`.
fn runtime_available(runtime: &str) -> bool {
match runtime {
"node" => crate::paths::look_path("npx").is_some(),
"python" => crate::paths::look_path("uvx").is_some(),
_ => true,
}
}
/// Le catalogue embarqué (celui du Go, `mcp_catalog.json`).
static CATALOG: LazyLock<Result<Vec<Value>, String>> = LazyLock::new(|| {
let raw = include_str!("../../../../internal/loki/mcp_catalog.json");
let v: Value =
serde_json::from_str(raw).map_err(|e| format!("catalogue MCP illisible : {e}"))?;
Ok(v.get("servers")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default()
.into_iter()
.map(|mut e| {
// Les champs du type Go, dans son ordre (le reste est ignoré).
let mut o = Map::new();
for (k, def) in [
("id", json!("")),
("name", json!("")),
("category", json!("")),
("description", json!("")),
("homepage", json!("")),
("runtime", json!("")),
("command", json!("")),
("args", json!([])),
("env", json!({})),
("secrets", json!([])),
] {
o.insert(k.into(), e.get_mut(k).map(Value::take).unwrap_or(def));
}
Value::Object(o)
})
.collect())
});
fn catalog_entries() -> Result<&'static Vec<Value>, String> {
CATALOG.as_ref().map_err(Clone::clone)
}
// ---------------------------------------------------------------------------
// Routes
// ---------------------------------------------------------------------------
/// Les routes servies par ce module.
pub fn handles(path: &str) -> bool {
matches!(
path,
"/api/mcp"
| "/api/mcp/catalog"
| "/api/mcp/save"
| "/api/mcp/delete"
| "/api/mcp/toggle"
| "/api/mcp/tool"
| "/api/mcp/test"
)
}
fn fail(code: StatusCode, e: impl Into<String>) -> Response<Body> {
send_json(code, &json!({ "ok": false, "error": e.into() }))
}
const SAVE_REQ_TYPE: &str = "mcpSaveReq";
const NAME_REQ_TYPE: &str = r#"struct { Name string "json:\"name\"" }"#;
const TOGGLE_REQ_TYPE: &str = r#"struct { Name string "json:\"name\""; On bool "json:\"on\"" }"#;
const TOOL_REQ_TYPE: &str =
r#"struct { Name string "json:\"name\""; Tool string "json:\"tool\""; On bool "json:\"on\"" }"#;
async fn read_body(body: Incoming) -> Vec<u8> {
Limited::new(body, 1 << 20)
.collect()
.await
.map(|b| b.to_bytes().to_vec())
.unwrap_or_default()
}
/// La version du Go (annoncée aux serveurs dans `clientInfo`).
async fn go_version(ctx: &Ctx) -> String {
ctx.internal
.auth()
.await
.map(|i| i.version)
.unwrap_or_default()
}
/// `from_go` : remise par le Go (tunnel), déjà authentifiée par lui.
pub async fn serve(
ctx: &Ctx,
mcp: &Arc<Mcp>,
req: Request<Incoming>,
from_go: bool,
) -> Response<Body> {
let (parts, body) = req.into_parts();
let version = if from_go {
go_version(ctx).await
} else {
match ctx.guard(&parts.headers).await {
Ok(v) => v,
Err(refus) => return *refus,
}
};
let store = &ctx.store;
let servers_ok =
|list: Vec<Value>| send_json(StatusCode::OK, &json!({ "ok": true, "servers": list }));
match parts.uri.path() {
"/api/mcp" => servers_ok(mcp.status(store, &version).await),
"/api/mcp/catalog" => match mcp.catalog(store).await {
Ok(c) => send_json(StatusCode::OK, &json!({ "ok": true, "catalog": c })),
Err(e) => fail(StatusCode::INTERNAL_SERVER_ERROR, e),
},
"/api/mcp/save" => {
let raw = read_body(body).await;
let d = gojson::decode_struct(
&raw,
SAVE_REQ_TYPE,
&[
("name", Kind::Str),
("command", Kind::Str),
("args", Kind::StrList),
("env", Kind::StrMap),
("url", Kind::Str),
("headers", Kind::StrMap),
("enabled", Kind::OptBool),
],
);
if let Some(e) = d.err.clone() {
return fail(StatusCode::BAD_REQUEST, e);
}
let cfg = ServerConfig {
command: d.str("command").trim().to_string(),
args: d.str_list("args").unwrap_or_default(),
env: d.str_map("env").unwrap_or_default().into_iter().collect(),
url: d.str("url").trim().to_string(),
headers: d
.str_map("headers")
.unwrap_or_default()
.into_iter()
.collect(),
enabled: d.opt_bool("enabled").unwrap_or(true),
disabled_tools: Vec::new(),
};
let name = d.str("name");
let res = mcp
.mutate(store, name.trim(), |s| config::set_server(s, &name, cfg))
.await;
if let Err(e) = res {
return fail(StatusCode::BAD_REQUEST, e);
}
servers_ok(mcp.status(store, &version).await)
}
"/api/mcp/delete" => {
let raw = read_body(body).await;
let d = gojson::decode_struct(&raw, NAME_REQ_TYPE, &[("name", Kind::Str)]);
if let Some(e) = d.err.clone() {
return fail(StatusCode::BAD_REQUEST, e);
}
let name = d.str("name");
if let Err(e) = mcp
.mutate(store, &name, |s| config::delete_server(s, &name))
.await
{
return fail(StatusCode::BAD_REQUEST, e);
}
send_json(StatusCode::OK, &json!({ "ok": true }))
}
"/api/mcp/toggle" => {
let raw = read_body(body).await;
let d = gojson::decode_struct(
&raw,
TOGGLE_REQ_TYPE,
&[("name", Kind::Str), ("on", Kind::Bool)],
);
if let Some(e) = d.err.clone() {
return fail(StatusCode::BAD_REQUEST, e);
}
let (name, on) = (d.str("name"), d.bool("on"));
if let Err(e) = mcp
.mutate(store, &name, |s| config::set_enabled(s, &name, on))
.await
{
return fail(StatusCode::BAD_REQUEST, e);
}
servers_ok(mcp.status(store, &version).await)
}
"/api/mcp/tool" => {
let raw = read_body(body).await;
let d = gojson::decode_struct(
&raw,
TOOL_REQ_TYPE,
&[("name", Kind::Str), ("tool", Kind::Str), ("on", Kind::Bool)],
);
if let Some(e) = d.err.clone() {
return fail(StatusCode::BAD_REQUEST, e);
}
let (name, tool, on) = (d.str("name"), d.str("tool"), d.bool("on"));
if let Err(e) = mcp
.mutate(store, &name, |s| {
config::set_tool_enabled(s, &name, &tool, on)
})
.await
{
return fail(StatusCode::BAD_REQUEST, e);
}
servers_ok(mcp.status(store, &version).await)
}
"/api/mcp/test" => {
let raw = read_body(body).await;
let d = gojson::decode_struct(&raw, NAME_REQ_TYPE, &[("name", Kind::Str)]);
if let Some(e) = d.err.clone() {
return fail(StatusCode::BAD_REQUEST, e);
}
mcp.invalidate(&d.str("name")).await;
servers_ok(mcp.status(store, &version).await)
}
_ => fail(StatusCode::NOT_FOUND, "route inconnue"),
}
}
/// L'API interne de la façade pour le Go (jeton interne vérifié par
/// l'appelant) : outils annoncés, appels, pré-chauffage.
pub async fn serve_internal(ctx: &Ctx, mcp: &Arc<Mcp>, req: Request<Incoming>) -> Response<Body> {
let path = req.uri().path().to_string();
let raw = read_body(req.into_body()).await;
let version = go_version(ctx).await;
match path.as_str() {
INTERNAL_TOOLS => send_json(
StatusCode::OK,
&json!({ "tools": mcp.tools(&ctx.store, &version).await }),
),
INTERNAL_CALL => {
let v: Value = serde_json::from_slice(&raw).unwrap_or_default();
let name = v.get("name").and_then(Value::as_str).unwrap_or("");
let args = v
.get("args")
.and_then(Value::as_object)
.cloned()
.unwrap_or_default();
let out = mcp.call(&ctx.store, name, &args, &version).await;
send_json(StatusCode::OK, &json!({ "result": out }))
}
INTERNAL_PREWARM => {
mcp.prewarm(ctx.store.clone(), version);
send_json(StatusCode::OK, &json!({ "ok": true }))
}
_ => fail(StatusCode::NOT_FOUND, "route inconnue"),
}
}
/// Les chemins de l'API interne de ce module.
pub fn is_internal(path: &str) -> bool {
matches!(path, INTERNAL_TOOLS | INTERNAL_CALL | INTERNAL_PREWARM)
}
+15
View File
@@ -87,6 +87,9 @@ pub struct Config {
pub system: Option<Arc<crate::system::System>>,
/// Hub des projets tenu par la façade (`LOKI_PROJECTS=rust`).
pub projects: bool,
/// Serveurs MCP (pool de sessions, configuration, catalogue) tenus par la
/// façade (`LOKI_MCP=rust`) ; `None` : au Go.
pub mcp: Option<Arc<crate::mcp::Mcp>>,
}
struct Proxy {
@@ -124,6 +127,7 @@ pub async fn serve(
engine_ctl: cfg.engine_ctl,
system: cfg.system,
projects: cfg.projects,
mcp: cfg.mcp,
internal_token: cfg.internal_token.clone(),
},
});
@@ -192,6 +196,12 @@ impl Proxy {
d.shutdown_fast().await;
return plain(StatusCode::NO_CONTENT, "");
}
if from_go
&& let Some(m) = &self.ctx.mcp
&& crate::mcp::is_internal(path)
{
return crate::mcp::serve_internal(&self.ctx, m, req).await;
}
// L'API interne du Go n'est jamais joignable de l'extérieur.
if path.starts_with("/internal/") {
return plain(StatusCode::NOT_FOUND, "404 page not found");
@@ -216,6 +226,11 @@ impl Proxy {
{
return chatapi::serve(&self.ctx, chat, req, from_go).await;
}
if let Some(m) = &self.ctx.mcp
&& crate::mcp::handles(path)
{
return crate::mcp::serve(&self.ctx, m, req, from_go).await;
}
if self.ctx.projects && crate::projects_api::handles(path) {
return crate::projects_api::serve(&self.ctx, req, from_go).await;
}
+2
View File
@@ -633,6 +633,7 @@ pub struct Opts {
pub engine_ctl: Option<Arc<loki_rs::enginectl::EngineCtl>>,
pub system: Option<Arc<loki_rs::system::System>>,
pub projects: bool,
pub mcp: Option<Arc<loki_rs::mcp::Mcp>>,
}
/// Comme [`start`], avec la conversation tenue par la façade
@@ -732,6 +733,7 @@ pub async fn start_with(
engine_ctl: opts.engine_ctl,
system: opts.system,
projects: opts.projects,
mcp: opts.mcp,
};
tokio::spawn(loki_rs::proxy::serve(fl, cfg, async move {
let _ = rx.await;
+1
View File
@@ -104,6 +104,7 @@ async fn banc(cle: Option<&str>) -> Banc {
engine_ctl: None,
system: None,
projects: false,
mcp: None,
};
tokio::spawn(loki_rs::proxy::serve(l, cfg, async move {
let _ = rx.await;
+651
View File
@@ -0,0 +1,651 @@
//! Lot 20 : serveurs MCP tenus par la façade (`LOKI_MCP=rust`).
//!
//! Le journal tiré du Go (`internal/loki/parity_mcp_test.go` →
//! `tests/parity_mcp/mcp.json`) est rejoué : fonctions pures, lecture et
//! mutations de la configuration, puis un faux serveur MCP (Streamable HTTP)
//! qui resert ce qu'un vrai serveur du SDK Go a mis sur le fil — l'état, les
//! outils annoncés au modèle et les résultats d'appel tirés par la façade
//! doivent être ceux du client Go. Le transport stdio, les routes de
//! configuration et l'API interne pour le Go sont vérifiés de bout en bout.
mod common;
use std::collections::BTreeMap;
use std::convert::Infallible;
use std::sync::{Arc, Mutex};
use bytes::Bytes;
use common::fakego;
use http_body_util::{BodyExt, Full};
use hyper::service::service_fn;
use hyper::{Request, Response};
use hyper_util::rt::TokioIo;
use loki_rs::mcp::{self, Mcp, config as cf};
use serde_json::{Map, Value, json};
use tokio::net::TcpListener;
fn fixture() -> Value {
serde_json::from_str(include_str!("parity_mcp/mcp.json")).expect("journal")
}
fn s(v: &Value) -> &str {
v.as_str().unwrap()
}
#[test]
fn parite_des_fonctions_pures() {
let fx = fixture();
for (k, v) in fx["sanitize"].as_object().unwrap() {
assert_eq!(cf::sanitize(k), s(v), "{k:?}");
}
for c in fx["schemas"].as_array().unwrap() {
let v: Value = serde_json::from_str(s(&c["in"])).unwrap();
assert_eq!(cf::normalize_schema(Some(&v)), c["out"], "{}", c["in"]);
}
assert_eq!(
cf::normalize_schema(None),
json!({"type":"object","properties":{}})
);
for c in fx["labels"].as_array().unwrap() {
let m: Map<String, Value> = serde_json::from_str(s(&c["in"])).unwrap();
assert_eq!(cf::arg_label(&m), s(&c["out"]), "{}", c["in"]);
}
for c in fx["tool_server"].as_array().unwrap() {
let got = cf::tool_server(s(&c["name"]), s(&c["desc"]));
assert_eq!(got.is_some(), c["ok"].as_bool().unwrap(), "{c}");
assert_eq!(got.unwrap_or_default(), s(&c["server"]), "{c}");
}
for c in fx["prompt_line"].as_array().unwrap() {
let tools: Vec<(String, String)> = c["tools"]
.as_array()
.map(|a| {
a.iter()
.map(|p| (s(&p[0]).to_string(), s(&p[1]).to_string()))
.collect()
})
.unwrap_or_default();
assert_eq!(cf::prompt_line(&tools), s(&c["out"]), "{c}");
}
for (cmd, v) in fx["timeouts"].as_object().unwrap() {
let cfg = cf::ServerConfig {
command: cmd.clone(),
..Default::default()
};
assert_eq!(
cf::timeout_for(&cfg).as_secs() as i64,
v["secs"].as_i64().unwrap(),
"{cmd}"
);
assert_eq!(cf::timeout_error(&cfg), s(&v["err"]), "{cmd}");
}
for c in fx["flatten"].as_array().unwrap() {
let v: Value = serde_json::from_str(s(&c["in"])).unwrap();
assert_eq!(cf::flatten_content(&v), s(&c["out"]), "{}", c["in"]);
}
}
#[test]
fn parite_de_la_configuration() {
let fx = fixture();
for c in fx["decode"].as_array().unwrap() {
let m = cf::decode_servers(Some(s(&c["in"]).as_bytes()));
let out = if s(&c["in"]) == "null" && m.is_empty() {
"null".to_string()
} else {
loki_rs::gojson::marshal(&m)
};
assert_eq!(out, s(&c["out"]), "{}", c["in"]);
for (n, want) in c["transports"].as_object().unwrap() {
let cfg = &m[n];
let got = format!(
"{}|{}",
cfg.transport(),
cfg.validate().err().unwrap_or_default()
);
assert_eq!(got, s(want), "{n}");
}
}
let mut servers = cf::Servers::new();
let mut saved = false;
for m in fx["mutations"].as_array().unwrap() {
let name = s(&m["name"]);
let res = match s(&m["op"]) {
"set" => {
let cfg: cf::ServerConfig = serde_json::from_str(s(&m["cfg"])).unwrap();
cf::set_server(&mut servers, name, cfg)
}
"enable" => cf::set_enabled(&mut servers, name, m["on"].as_bool().unwrap_or(false)),
"tool" => cf::set_tool_enabled(
&mut servers,
name,
s(&m["tool"]),
m["on"].as_bool().unwrap_or(false),
),
"delete" => cf::delete_server(&mut servers, name),
op => panic!("{op}"),
};
saved |= res.is_ok();
assert_eq!(res.err().unwrap_or_default(), s(&m["error"]), "{m}");
let state = if saved {
loki_rs::gojson::marshal(&servers)
} else {
String::new()
};
assert_eq!(state, s(&m["state"]), "{m}");
}
}
/// Un faux serveur MCP en Streamable HTTP : resert ce que le serveur du SDK
/// Go a répondu. `tools/call` en flux SSE, le reste en JSON.
async fn fake_server(wire: Value, seen: Arc<Mutex<Vec<String>>>) -> String {
let l = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = l.local_addr().unwrap();
tokio::spawn(async move {
while let Ok((st, _)) = l.accept().await {
let (wire, seen) = (wire.clone(), seen.clone());
tokio::spawn(async move {
let svc = service_fn(move |req: Request<hyper::body::Incoming>| {
let (wire, seen) = (wire.clone(), seen.clone());
async move {
let sid = req
.headers()
.get("mcp-session-id")
.map(|v| v.to_str().unwrap().to_string());
let auth = req
.headers()
.get("x-cle")
.map(|v| v.to_str().unwrap().to_string());
let method = req.method().clone();
let b = req.into_body().collect().await.unwrap().to_bytes();
let mut r = Response::new(Full::new(Bytes::new()));
if method == hyper::Method::DELETE {
seen.lock().unwrap().push("DELETE".into());
return Ok::<_, Infallible>(r);
}
let v: Value = serde_json::from_slice(&b).unwrap();
let m = v["method"].as_str().unwrap_or("").to_string();
seen.lock().unwrap().push(format!(
"{m} {} {}",
sid.as_deref().unwrap_or("-"),
auth.as_deref().unwrap_or("-")
));
let result = match m.as_str() {
"initialize" => {
r.headers_mut()
.insert("Mcp-Session-Id", "sess-1".parse().unwrap());
json!({ "protocolVersion": "2025-06-18",
"capabilities": { "tools": {} },
"serverInfo": { "name": "faux", "version": "1" } })
}
"tools/list" => wire["tools/list"].clone(),
"tools/call" => {
let name = v["params"]["name"].as_str().unwrap();
let res = wire["tools/call"][name].clone();
let body = format!(
"event: message\ndata: {}\n\n",
json!({ "jsonrpc": "2.0", "id": v["id"], "result": res })
);
r.headers_mut()
.insert("content-type", "text/event-stream".parse().unwrap());
*r.body_mut() = Full::new(Bytes::from(body));
return Ok(r);
}
_ => {
*r.status_mut() = hyper::StatusCode::ACCEPTED;
return Ok(r);
}
};
r.headers_mut()
.insert("content-type", "application/json".parse().unwrap());
*r.body_mut() = Full::new(Bytes::from(
json!({ "jsonrpc": "2.0", "id": v["id"], "result": result })
.to_string(),
));
Ok(r)
}
});
let _ = hyper::server::conn::http1::Builder::new()
.serve_connection(TokioIo::new(st), svc)
.await;
});
}
});
format!("http://{addr}")
}
async fn internal(facade: std::net::SocketAddr, path: &str, body: Value) -> Value {
let (code, b) = fakego::send(
facade,
"POST",
path,
&[(loki_rs::auth::INTERNAL_HEADER, fakego::JETON)],
&body.to_string(),
)
.await;
assert_eq!(code, 200, "{path}");
serde_json::from_slice(&b).unwrap()
}
#[tokio::test]
async fn parite_contre_le_serveur_du_sdk() {
let fx = fixture();
let seen = Arc::new(Mutex::new(Vec::new()));
let url = fake_server(fx["wire"].clone(), seen.clone()).await;
let tmp = tempfile::tempdir().unwrap();
let pool = Arc::new(Mcp::new(tmp.path().to_path_buf()));
let (facade, go, _stop, _) = fakego::start_with(
tmp.path().to_path_buf(),
fakego::Opts {
mcp: Some(pool.clone()),
..Default::default()
},
)
.await;
go.set("state", "mcp", &s(&fx["store"]).replace("{{URL}}", &url));
let mask = |v: &Value| -> Value {
serde_json::from_str(&v.to_string().replace(&url, "{{URL}}")).unwrap()
};
let (_, b) = fakego::send(facade, "GET", "/api/mcp", &[], "").await;
let v: Value = serde_json::from_slice(&b).unwrap();
assert_eq!(v["ok"], true);
assert_eq!(mask(&v["servers"]), fx["status"]);
let tools = internal(facade, mcp::INTERNAL_TOOLS, json!({})).await;
assert_eq!(tools["tools"], fx["tools"]);
let pairs: Vec<(String, String)> = tools["tools"]
.as_array()
.unwrap()
.iter()
.map(|t| {
(
s(&t["function"]["name"]).to_string(),
s(&t["function"]["description"]).to_string(),
)
})
.collect();
assert_eq!(cf::prompt_line(&pairs), s(&fx["prompt_line_live"]));
for (name, want) in fx["calls"].as_object().unwrap() {
let args = if name == "mcp__alpha__echo" {
json!({ "text": "salut <toi>" })
} else {
json!({})
};
let r = internal(
facade,
mcp::INTERNAL_CALL,
json!({ "name": name, "args": args }),
)
.await;
assert_eq!(r["result"], *want, "{name}");
}
// Une seule session, partagée par l'état, les outils et les appels.
let seen = seen.lock().unwrap().clone();
assert_eq!(
seen.iter().filter(|l| l.starts_with("initialize")).count(),
1,
"{seen:?}"
);
assert!(seen.iter().any(|l| l == "tools/call sess-1 -"), "{seen:?}");
}
/// Un faux serveur MCP en stdio : sh, un message par ligne.
fn stdio_server(dir: &std::path::Path) -> String {
let p = dir.join("faux-mcp");
let body = r#"#!/bin/sh
while IFS= read -r line; do
id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\).*/\1/p')
case "$line" in
*'"method":"initialize"'*)
printf '{"jsonrpc":"2.0","method":"notifications/message","params":{}}\n'
printf '{"jsonrpc":"2.0","id":%s,"result":{"protocolVersion":"2025-06-18","capabilities":{},"serverInfo":{"name":"sh","version":"1"}}}\n' "$id" ;;
*'"method":"tools/list"'*)
printf '{"jsonrpc":"2.0","id":99,"method":"ping"}\n'
printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[{"name":"date","description":"Heure","inputSchema":{"type":"object"}},{"name":"crash","inputSchema":{"type":"object"}}]}}\n' "$id" ;;
*'"name":"crash"'*) exit 3 ;;
*'"method":"tools/call"'*)
printf '{"jsonrpc":"2.0","id":%s,"result":{"content":[{"type":"text","text":"%s"}]}}\n' "$id" "$FAUX_ENV" ;;
*'"result":{}'*) echo "pong reçu" >> "$TRACE" ;;
esac
done
"#;
let ok = std::process::Command::new("sh")
.args(["-c", "printf %s \"$1\" > \"$2\" && chmod 755 \"$2\""])
.arg("sh")
.arg(body)
.arg(&p)
.status()
.unwrap()
.success();
assert!(ok);
p.to_string_lossy().into_owned()
}
#[tokio::test]
async fn transport_stdio_et_reconnexion() {
let tmp = tempfile::tempdir().unwrap();
let bin = stdio_server(tmp.path());
let trace = tmp.path().join("trace");
let pool = Arc::new(Mcp::new(tmp.path().to_path_buf()));
let (facade, go, _stop, _) = fakego::start_with(
tmp.path().to_path_buf(),
fakego::Opts {
mcp: Some(pool.clone()),
..Default::default()
},
)
.await;
let mut env = BTreeMap::new();
env.insert("FAUX_ENV".to_string(), "valeur d'env".to_string());
env.insert("TRACE".to_string(), trace.to_string_lossy().into_owned());
let mut servers = cf::Servers::new();
servers.insert(
"sh".into(),
cf::ServerConfig {
command: bin,
env,
..Default::default()
},
);
go.set("state", "mcp", &loki_rs::gojson::marshal(&servers));
let (_, b) = fakego::send(facade, "GET", "/api/mcp", &[], "").await;
let v: Value = serde_json::from_slice(&b).unwrap();
assert_eq!(v["servers"][0]["connected"], true, "{v}");
assert_eq!(v["servers"][0]["tools"], json!(["date", "crash"]));
let r = internal(
facade,
mcp::INTERNAL_CALL,
json!({ "name": "mcp__sh__date", "args": {} }),
)
.await;
assert_eq!(r["result"], "valeur d'env");
// Le serveur meurt en plein appel : une reconnexion, puis l'erreur.
let r = internal(
facade,
mcp::INTERNAL_CALL,
json!({ "name": "mcp__sh__crash", "args": {} }),
)
.await;
let out = s(&r["result"]);
assert!(
out.starts_with("[erreur MCP] calling \"tools/call\""),
"{out}"
);
// Après la reconnexion, le serveur répond de nouveau.
let r = internal(
facade,
mcp::INTERNAL_CALL,
json!({ "name": "mcp__sh__date", "args": {} }),
)
.await;
assert_eq!(r["result"], "valeur d'env");
// Le ping du serveur a reçu sa réponse.
let t = std::fs::read_to_string(&trace).unwrap_or_default();
assert!(t.contains("pong reçu"), "{t:?}");
pool.close_all().await;
}
#[tokio::test]
async fn routes_de_configuration() {
let tmp = tempfile::tempdir().unwrap();
let seen = Arc::new(Mutex::new(Vec::new()));
let fx = fixture();
let url = fake_server(fx["wire"].clone(), seen.clone()).await;
let pool = Arc::new(Mcp::new(tmp.path().to_path_buf()));
let (facade, go, _stop, _) = fakego::start_with(
tmp.path().to_path_buf(),
fakego::Opts {
mcp: Some(pool.clone()),
..Default::default()
},
)
.await;
let post = |path: &'static str, body: String| async move {
let (c, b) = fakego::send(facade, "POST", path, &[], &body).await;
(c.as_u16(), serde_json::from_slice::<Value>(&b).unwrap())
};
// Ajout : actif par défaut, en-têtes envoyés, état rendu.
let (c, v) = post(
"/api/mcp/save",
json!({ "name": " web ", "url": format!(" {url} "), "headers": { "X-Cle": "k" } })
.to_string(),
)
.await;
assert_eq!(c, 200, "{v}");
assert_eq!(v["servers"][0]["name"], "web");
assert_eq!(v["servers"][0]["connected"], true);
assert!(
seen.lock().unwrap().iter().any(|l| l == "initialize - k"),
"{:?}",
seen.lock().unwrap()
);
let stored: Value = serde_json::from_str(&go.get("state", "mcp").unwrap()).unwrap();
assert_eq!(stored["web"]["url"], json!(url));
// Erreurs de la configuration et du JSON.
let (c, v) = post("/api/mcp/save", r#"{"name":"a__b","command":"x"}"#.into()).await;
assert_eq!(
(c, v["error"].clone()),
(
400,
json!("le nom ne peut pas contenir '__' (réservé au namespacing des outils)")
)
);
let (c, v) = post(
"/api/mcp/save",
r#"{"name":"x","args":"pas une liste"}"#.into(),
)
.await;
assert_eq!(
(c, v["error"].clone()),
(
400,
json!(
"json: cannot unmarshal string into Go struct field mcpSaveReq.args of type []string"
)
)
);
// Un outil masqué : connu de l'état, absent des outils annoncés.
let (c, v) = post(
"/api/mcp/tool",
r#"{"name":"web","tool":"secret","on":false}"#.into(),
)
.await;
assert_eq!(
(c, v["servers"][0]["disabled"].clone()),
(200, json!(["secret"]))
);
let tools = internal(facade, mcp::INTERNAL_TOOLS, json!({})).await;
assert!(
!tools["tools"]
.as_array()
.unwrap()
.iter()
.any(|t| t["function"]["name"] == "mcp__web__secret")
);
// Éteint : plus de connexion, plus d'outils ; la session est fermée.
let (c, v) = post("/api/mcp/toggle", r#"{"name":"web","on":false}"#.into()).await;
assert_eq!(
(c, v["servers"][0]["connected"].clone()),
(200, json!(false))
);
assert!(seen.lock().unwrap().iter().any(|l| l == "DELETE"));
let tools = internal(facade, mcp::INTERNAL_TOOLS, json!({})).await;
assert_eq!(tools["tools"], json!([]));
let (c, v) = post("/api/mcp/toggle", r#"{"name":"nope","on":true}"#.into()).await;
assert_eq!(
(c, v["error"].clone()),
(400, json!("serveur MCP inconnu: nope"))
);
// Tester : reconnexion.
post("/api/mcp/toggle", r#"{"name":"web","on":true}"#.into()).await;
let before = seen
.lock()
.unwrap()
.iter()
.filter(|l| l.starts_with("initialize"))
.count();
let (c, _) = post("/api/mcp/test", r#"{"name":"web"}"#.into()).await;
assert_eq!(c, 200);
let after = seen
.lock()
.unwrap()
.iter()
.filter(|l| l.starts_with("initialize"))
.count();
assert_eq!(after, before + 1);
// Suppression.
let (c, v) = post("/api/mcp/delete", r#"{"name":"web"}"#.into()).await;
assert_eq!((c, v), (200, json!({"ok": true})));
assert_eq!(go.get("state", "mcp").unwrap(), "{}");
let (c, v) = post("/api/mcp/delete", r#"{"name":"web"}"#.into()).await;
assert_eq!(
(c, v["error"].clone()),
(400, json!("serveur MCP inconnu: web"))
);
pool.close_all().await;
}
#[tokio::test]
async fn catalogue_embarque() {
let tmp = tempfile::tempdir().unwrap();
let pool = Arc::new(Mcp::new(tmp.path().to_path_buf()));
let (facade, go, _stop, _) = fakego::start_with(
tmp.path().to_path_buf(),
fakego::Opts {
mcp: Some(pool),
..Default::default()
},
)
.await;
go.set("state", "mcp", r#"{"filesystem":{"command":"npx"}}"#);
let (_, b) = fakego::send(facade, "GET", "/api/mcp/catalog", &[], "").await;
let v: Value = serde_json::from_slice(&b).unwrap();
let cat = v["catalog"].as_array().unwrap();
assert!(cat.len() > 10);
let fs = cat.iter().find(|e| e["id"] == "filesystem").unwrap();
assert_eq!(fs["installed"], true);
let ws = format!("{}/workspace", tmp.path().display());
assert!(fs["args"].as_array().unwrap().contains(&json!(ws)), "{fs}");
assert!(
fs.get("env").is_none() && fs.get("secrets").is_none(),
"{fs}"
);
assert!(cat.iter().all(|e| e.get("runtimeOk").is_some()));
assert!(cat.iter().filter(|e| e["installed"] == true).count() == 1);
}
#[tokio::test]
async fn sans_la_variable_mcp_reste_au_go() {
let tmp = tempfile::tempdir().unwrap();
let (facade, _go, _stop) = fakego::start(tmp.path().to_path_buf()).await;
for p in [
"/api/mcp",
"/api/mcp/save",
"/api/mcp/catalog",
"/api/mcp/test",
] {
let (_, _, b) = fakego::get(facade, p, &[]).await;
assert_eq!(&b[..], format!("go:{p}").as_bytes());
}
// L'API interne MCP de la façade n'existe pas sans le lot.
let (c, _) = fakego::send(
facade,
"POST",
mcp::INTERNAL_TOOLS,
&[(loki_rs::auth::INTERNAL_HEADER, fakego::JETON)],
"{}",
)
.await;
assert_eq!(c, 404);
}
/// Processus vivants dont la ligne de commande contient `needle`.
fn alive(needle: &str) -> usize {
std::fs::read_dir("/proc")
.unwrap()
.flatten()
.filter(|e| {
e.file_name()
.to_string_lossy()
.bytes()
.all(|b| b.is_ascii_digit())
})
.filter(|e| {
std::fs::read(e.path().join("cmdline"))
.map(|b| String::from_utf8_lossy(&b).contains(needle))
.unwrap_or(false)
})
.filter(|e| {
// Un zombie n'est plus un serveur vivant.
std::fs::read_to_string(e.path().join("stat"))
.map(|s| !s.contains(") Z "))
.unwrap_or(false)
})
.count()
}
#[tokio::test]
async fn session_invalidee_pendant_la_connexion() {
let tmp = tempfile::tempdir().unwrap();
let p = tmp.path().join("lent-mcp");
let body = r#"#!/bin/sh
while IFS= read -r line; do
id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\).*/\1/p')
case "$line" in
*'"method":"initialize"'*)
sleep 1
printf '{"jsonrpc":"2.0","id":%s,"result":{"protocolVersion":"2025-06-18","capabilities":{},"serverInfo":{"name":"lent","version":"1"}}}\n' "$id" ;;
*'"method":"tools/list"'*)
printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[{"name":"t","inputSchema":{"type":"object"}}]}}\n' "$id" ;;
esac
done
"#;
assert!(
std::process::Command::new("sh")
.args(["-c", "printf %s \"$1\" > \"$2\" && chmod 755 \"$2\""])
.arg("sh")
.arg(body)
.arg(&p)
.status()
.unwrap()
.success()
);
let bin = p.to_string_lossy().into_owned();
let pool = Arc::new(Mcp::new(tmp.path().to_path_buf()));
let (facade, go, _stop, _) = fakego::start_with(
tmp.path().to_path_buf(),
fakego::Opts {
mcp: Some(pool.clone()),
..Default::default()
},
)
.await;
let mut servers = cf::Servers::new();
servers.insert(
"lent".into(),
cf::ServerConfig {
command: bin.clone(),
..Default::default()
},
);
go.set("state", "mcp", &loki_rs::gojson::marshal(&servers));
// Une première connexion part, puis le serveur est invalidé avant qu'elle
// aboutisse (« tester ») : elle ne doit jamais être republiée.
let first = tokio::spawn(async move { fakego::send(facade, "GET", "/api/mcp", &[], "").await });
tokio::time::sleep(std::time::Duration::from_millis(300)).await;
let (c, b) = fakego::send(facade, "POST", "/api/mcp/test", &[], r#"{"name":"lent"}"#).await;
assert_eq!(c, 200);
let v: Value = serde_json::from_slice(&b).unwrap();
assert_eq!(v["servers"][0]["connected"], true);
let _ = first.await;
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
assert_eq!(
alive(&bin),
1,
"un seul serveur vivant après l'invalidation"
);
pool.close_all().await;
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
assert_eq!(alive(&bin), 0);
}
+760
View File
@@ -0,0 +1,760 @@
{
"calls": {
"mcp__alpha__big": "éééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééLine truncated
"mcp__alpha__dot_tool": "point",
"mcp__alpha__echo": "salut <toi>",
"mcp__alpha__empty": "[ok] (aucune sortie)",
"mcp__alpha__fail": "[l'outil a renvoyé une erreur]\nboum",
"mcp__alpha__link": "{\"type\":\"resource_link\",\"mimeType\":\"text/plain\",\"uri\":\"file:///a.txt\",\"name\":\"a.txt\",\"title\":\"A\",\"size\":12}",
"mcp__alpha__media": "[image image/png]\n[audio audio/wav]",
"mcp__alpha__multi": "un\ndeux",
"mcp__alpha__nope": "[erreur] outil MCP inconnu ou serveur non connecté: mcp__alpha__nope",
"mcp__alpha__secret": "chut",
"mcp__alpha__struct": "{\n \"a\": \"\\u003cx\\u003e \\u0026 y\",\n \"b\": [\n 1,\n 2\n ]\n}",
"mcp__alpha__titled": "ok",
"mcp__zeta__x": "[erreur] outil MCP inconnu ou serveur non connecté: mcp__zeta__x"
},
"decode": [
{
"in": "{\"a\":{\"command\":\"npx\",\"args\":[\"-y\",\"x\"],\"env\":{\"K\":\"v\"}}}",
"out": "{\"a\":{\"command\":\"npx\",\"args\":[\"-y\",\"x\"],\"env\":{\"K\":\"v\"},\"enabled\":true}}",
"transports": {
"a": "stdio|"
}
},
{
"in": "{\"b\":{\"url\":\"http://h/mcp\",\"headers\":{\"Authorization\":\"Bearer t\"},\"enabled\":false,\"disabledTools\":[\"t1\"]}}",
"out": "{\"b\":{\"url\":\"http://h/mcp\",\"headers\":{\"Authorization\":\"Bearer t\"},\"enabled\":false,\"disabledTools\":[\"t1\"]}}",
"transports": {
"b": "http|"
}
},
{
"in": "{\"c\":{\"command\":\"x\",\"enabled\":null}}",
"out": "{\"c\":{\"command\":\"x\",\"enabled\":false}}",
"transports": {
"c": "stdio|"
}
},
{
"in": "null",
"out": "null",
"transports": {}
},
{
"in": "{}",
"out": "{}",
"transports": {}
}
],
"flatten": [
{
"in": "{\"content\":[{\"type\":\"text\",\"text\":\" un \"},{\"type\":\"text\",\"text\":\"deux\\n\"}]}",
"out": "un \ndeux"
},
{
"in": "{\"content\":[{\"type\":\"image\",\"data\":\"cG5n\",\"mimeType\":\"image/png\"},{\"type\":\"audio\",\"data\":\"d2F2\",\"mimeType\":\"audio/wav\"}]}",
"out": "[image image/png]\n[audio audio/wav]"
},
{
"in": "{\"content\":[{\"type\":\"resource_link\",\"uri\":\"file:///a.txt\",\"name\":\"a.txt\",\"mimeType\":\"text/plain\",\"size\":12,\"title\":\"A <b>\"}]}",
"out": "{\"type\":\"resource_link\",\"mimeType\":\"text/plain\",\"uri\":\"file:///a.txt\",\"name\":\"a.txt\",\"title\":\"A \\u003cb\\u003e\",\"size\":12}"
},
{
"in": "{\"content\":[{\"type\":\"resource\",\"resource\":{\"uri\":\"file:///b\",\"mimeType\":\"text/plain\",\"text\":\"corps\"}}]}",
"out": "{\"type\":\"resource\",\"resource\":{\"uri\":\"file:///b\",\"mimeType\":\"text/plain\",\"text\":\"corps\"}}"
},
{
"in": "{\"content\":[],\"structuredContent\":{\"b\":[1,2],\"a\":\"<x> & y\"}}",
"out": "{\n \"a\": \"\\u003cx\\u003e \\u0026 y\",\n \"b\": [\n 1,\n 2\n ]\n}"
},
{
"in": "{\"content\":[{\"type\":\"text\",\"text\":\" \"}],\"structuredContent\":{\"k\":1}}",
"out": "{\n \"k\": 1\n}"
},
{
"in": "{\"content\":[{\"type\":\"text\",\"text\":\"t\"}],\"structuredContent\":{\"k\":1}}",
"out": "t"
},
{
"in": "{\"content\":[]}",
"out": ""
},
{
"in": "{\"content\":[{\"type\":\"text\",\"text\":\"boum\"}],\"isError\":true}",
"out": "boum"
}
],
"friendly_other": "autre",
"labels": [
{
"in": "{}",
"out": ""
},
{
"in": "{\"path\":\"/a\",\"query\":\"q\"}",
"out": "/a"
},
{
"in": "{\"query\":\"\",\"url\":\"http://x\"}",
"out": "http://x"
},
{
"in": "{\"name\":\"n\",\"command\":\"c\"}",
"out": "c"
},
{
"in": "{\"x\":1,\"y\":\"<b>\"}",
"out": "{\"x\":1,\"y\":\"\\u003cb\\u003e\"}"
},
{
"in": "{\"long\":\"éééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééé\"}",
"out": "{\"long\":\"ééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééé…"
},
{
"in": "{\"path\":3,\"file\":\"f\"}",
"out": "f"
},
{
"in": "{\"file\":\"f\",\"path\":\"p\"}",
"out": "p"
},
{
"in": "{\"moyen\":\"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\"}",
"out": "{\"moyen\":\"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa…"
}
],
"mutations": [
{
"op": "set",
"name": " fs ",
"cfg": "{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"<w>\"],\"env\":{\"A\":\"&\"}}",
"error": "",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true}}"
},
{
"op": "set",
"name": "",
"cfg": "{\"command\":\"x\"}",
"error": "nom de serveur vide",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true}}"
},
{
"op": "set",
"name": "a__b",
"cfg": "{\"command\":\"x\"}",
"error": "le nom ne peut pas contenir '__' (réservé au namespacing des outils)",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true}}"
},
{
"op": "set",
"name": "both",
"cfg": "{\"command\":\"x\",\"url\":\"http://u\"}",
"error": "un serveur MCP ne peut avoir à la fois 'command' (stdio) et 'url' (http)",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true}}"
},
{
"op": "set",
"name": "none",
"cfg": "{\"enabled\":true}",
"error": "un serveur MCP doit avoir soit 'command' (stdio) soit 'url' (http)",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true}}"
},
{
"op": "set",
"name": "web",
"cfg": "{\"url\":\"http://u/mcp\",\"headers\":{\"X\":\"1\"},\"enabled\":false}",
"error": "",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true},\"web\":{\"url\":\"http://u/mcp\",\"headers\":{\"X\":\"1\"},\"enabled\":false}}"
},
{
"op": "enable",
"name": "web",
"on": true,
"error": "",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true},\"web\":{\"url\":\"http://u/mcp\",\"headers\":{\"X\":\"1\"},\"enabled\":true}}"
},
{
"op": "enable",
"name": "nope",
"on": true,
"error": "serveur MCP inconnu: nope",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true},\"web\":{\"url\":\"http://u/mcp\",\"headers\":{\"X\":\"1\"},\"enabled\":true}}"
},
{
"op": "tool",
"name": "fs",
"tool": "write",
"error": "",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true,\"disabledTools\":[\"write\"]},\"web\":{\"url\":\"http://u/mcp\",\"headers\":{\"X\":\"1\"},\"enabled\":true}}"
},
{
"op": "tool",
"name": "fs",
"tool": "read",
"error": "",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true,\"disabledTools\":[\"write\",\"read\"]},\"web\":{\"url\":\"http://u/mcp\",\"headers\":{\"X\":\"1\"},\"enabled\":true}}"
},
{
"op": "tool",
"name": "fs",
"tool": "write",
"on": true,
"error": "",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true,\"disabledTools\":[\"read\"]},\"web\":{\"url\":\"http://u/mcp\",\"headers\":{\"X\":\"1\"},\"enabled\":true}}"
},
{
"op": "tool",
"name": "fs",
"tool": "read",
"error": "",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true,\"disabledTools\":[\"read\"]},\"web\":{\"url\":\"http://u/mcp\",\"headers\":{\"X\":\"1\"},\"enabled\":true}}"
},
{
"op": "tool",
"name": "nope",
"tool": "x",
"error": "serveur MCP inconnu: nope",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true,\"disabledTools\":[\"read\"]},\"web\":{\"url\":\"http://u/mcp\",\"headers\":{\"X\":\"1\"},\"enabled\":true}}"
},
{
"op": "delete",
"name": "web",
"error": "",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true,\"disabledTools\":[\"read\"]}}"
},
{
"op": "delete",
"name": "web",
"error": "serveur MCP inconnu: web",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true,\"disabledTools\":[\"read\"]}}"
},
{
"op": "delete",
"name": " fs",
"error": "serveur MCP inconnu: fs",
"state": "{\"fs\":{\"command\":\"npx\",\"args\":[\"-y\",\"@x/fs\",\"\\u003cw\\u003e\"],\"env\":{\"A\":\"\\u0026\"},\"enabled\":true,\"disabledTools\":[\"read\"]}}"
}
],
"prompt_line": [
{
"out": "",
"tools": []
},
{
"out": "",
"tools": [
[
"bash",
"shell"
]
]
},
{
"out": "\n\nMCP servers connected (their tools are prefixed mcp__<server>__<tool>): a (2), b (1).",
"tools": [
[
"mcp__b__x",
"[MCP: b] d"
],
[
"mcp__a__y",
"[MCP: a] d"
],
[
"mcp__a__z",
"[MCP: a] e"
]
]
},
{
"out": "\n\nMCP servers connected (their tools are prefixed mcp__<server>__<tool>): a]b (1).",
"tools": [
[
"mcp__a_b__x",
"[MCP: a]b] d"
]
]
}
],
"prompt_line_live": "\n\nMCP servers connected (their tools are prefixed mcp__<server>__<tool>): alpha (10).",
"sanitize": {
"": "",
"...": "",
"A-B_c9": "A-B_c9",
"__x__": "x",
"a.b/c": "a_b_c",
"alpha": "alpha",
"mon serveur!": "mon_serveur",
"é-t_é": "-t"
},
"schemas": [
{
"in": "null",
"out": {
"properties": {},
"type": "object"
}
},
{
"in": "{\"type\":\"object\",\"properties\":{\"a\":{\"type\":\"string\"}}}",
"out": {
"properties": {
"a": {
"type": "string"
}
},
"type": "object"
}
},
{
"in": "true",
"out": {
"properties": {},
"type": "object"
}
},
{
"in": "[1]",
"out": {
"properties": {},
"type": "object"
}
},
{
"in": "{}",
"out": {}
},
{
"in": "\"x\"",
"out": {
"properties": {},
"type": "object"
}
}
],
"status": [
{
"name": "alpha",
"transport": "http",
"enabled": true,
"connected": true,
"tools": [
"big",
"dot.tool",
"echo",
"empty",
"fail",
"link",
"media",
"multi",
"secret",
"struct",
"titled"
],
"disabled": [
"secret"
],
"url": "{{URL}}"
},
{
"name": "beta",
"transport": "stdio",
"enabled": true,
"connected": false,
"error": "exec: \"loki-mcp-absent-pour-test\": executable file not found in $PATH",
"tools": [],
"disabled": [],
"command": "loki-mcp-absent-pour-test"
},
{
"name": "delta",
"transport": "",
"enabled": true,
"connected": false,
"tools": [],
"disabled": []
},
{
"name": "gamma",
"transport": "http",
"enabled": false,
"connected": false,
"tools": [],
"disabled": [],
"url": "{{URL}}"
}
],
"store": "{\"alpha\":{\"url\":\"{{URL}}\",\"enabled\":true,\"disabledTools\":[\"secret\"]},\"beta\":{\"command\":\"loki-mcp-absent-pour-test\",\"enabled\":true},\"gamma\":{\"url\":\"{{URL}}\",\"enabled\":false},\"delta\":{\"enabled\":true}}",
"timeouts": {
"": {
"err": "délai dépassé — le serveur n'a pas répondu au handshake en 20s",
"secs": 20
},
"/usr/local/bin/NPX": {
"err": "délai dépassé — npx télécharge le paquet du serveur au premier lancement, ce qui peut être long ; clique « tester » pour réessayer (le paquet reste en cache), et vérifie que le conteneur atteint le registre (npm/PyPI)",
"secs": 180
},
"C:/bin/uvx.EXE": {
"err": "délai dépassé — uvx télécharge le paquet du serveur au premier lancement, ce qui peut être long ; clique « tester » pour réessayer (le paquet reste en cache), et vérifie que le conteneur atteint le registre (npm/PyPI)",
"secs": 180
},
"node": {
"err": "délai dépassé — le serveur n'a pas répondu au handshake en 20s",
"secs": 20
},
"npx": {
"err": "délai dépassé — npx télécharge le paquet du serveur au premier lancement, ce qui peut être long ; clique « tester » pour réessayer (le paquet reste en cache), et vérifie que le conteneur atteint le registre (npm/PyPI)",
"secs": 180
},
"npx.cmd": {
"err": "délai dépassé — npx télécharge le paquet du serveur au premier lancement, ce qui peut être long ; clique « tester » pour réessayer (le paquet reste en cache), et vérifie que le conteneur atteint le registre (npm/PyPI)",
"secs": 180
},
"uvx-x": {
"err": "délai dépassé — le serveur n'a pas répondu au handshake en 20s",
"secs": 20
},
"uvx.exe": {
"err": "délai dépassé — uvx télécharge le paquet du serveur au premier lancement, ce qui peut être long ; clique « tester » pour réessayer (le paquet reste en cache), et vérifie que le conteneur atteint le registre (npm/PyPI)",
"secs": 180
}
},
"tool_server": [
{
"desc": "[MCP: alpha] desc",
"name": "mcp__alpha__x",
"ok": true,
"server": "alpha"
},
{
"desc": "[MCP: a]b] desc",
"name": "mcp__a_b__x",
"ok": true,
"server": "a]b"
},
{
"desc": "[MCP: a] b] c] d",
"name": "mcp__a_b_c__x",
"ok": true,
"server": "a] b] c"
},
{
"desc": "pas de préfixe",
"name": "mcp__x__y",
"ok": false,
"server": ""
},
{
"desc": "[MCP: autre] d",
"name": "autre",
"ok": false,
"server": ""
},
{
"desc": "[MCP: a] d",
"name": "mcp__b__x",
"ok": false,
"server": ""
}
],
"tools": [
{
"type": "function",
"function": {
"name": "mcp__alpha__big",
"description": "[MCP: alpha] Long",
"parameters": {
"type": "object"
}
}
},
{
"type": "function",
"function": {
"name": "mcp__alpha__dot_tool",
"description": "[MCP: alpha] Nom pointé",
"parameters": {
"properties": {},
"type": "object"
}
}
},
{
"type": "function",
"function": {
"name": "mcp__alpha__echo",
"description": "[MCP: alpha] Répète le texte <tel quel>",
"parameters": {
"properties": {
"text": {
"type": "string"
}
},
"type": "object"
}
}
},
{
"type": "function",
"function": {
"name": "mcp__alpha__empty",
"description": "[MCP: alpha] Rien",
"parameters": {
"type": "object"
}
}
},
{
"type": "function",
"function": {
"name": "mcp__alpha__fail",
"description": "[MCP: alpha] Échoue",
"parameters": {
"type": "object"
}
}
},
{
"type": "function",
"function": {
"name": "mcp__alpha__link",
"description": "[MCP: alpha] Un lien",
"parameters": {
"type": "object"
}
}
},
{
"type": "function",
"function": {
"name": "mcp__alpha__media",
"description": "[MCP: alpha] Image et son",
"parameters": {
"type": "object"
}
}
},
{
"type": "function",
"function": {
"name": "mcp__alpha__multi",
"description": "[MCP: alpha] Deux blocs",
"parameters": {
"type": "object"
}
}
},
{
"type": "function",
"function": {
"name": "mcp__alpha__struct",
"description": "[MCP: alpha] Structuré",
"parameters": {
"type": "object"
}
}
},
{
"type": "function",
"function": {
"name": "mcp__alpha__titled",
"description": "[MCP: alpha] Titre seul",
"parameters": {
"type": "object"
}
}
}
],
"wire": {
"tools/call": {
"big": {
"content": [
{
"text": "éééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééééLine truncated
"type": "text"
}
]
},
"dot.tool": {
"content": [
{
"text": "point",
"type": "text"
}
]
},
"echo": {
"content": [
{
"text": "salut <toi>",
"type": "text"
}
]
},
"empty": {
"content": []
},
"fail": {
"content": [
{
"text": "boum",
"type": "text"
}
],
"isError": true
},
"link": {
"content": [
{
"mimeType": "text/plain",
"name": "a.txt",
"size": 12,
"title": "A",
"type": "resource_link",
"uri": "file:///a.txt"
}
]
},
"media": {
"content": [
{
"data": "cG5n",
"mimeType": "image/png",
"type": "image"
},
{
"data": "d2F2",
"mimeType": "audio/wav",
"type": "audio"
}
]
},
"multi": {
"content": [
{
"text": "un",
"type": "text"
},
{
"text": "deux",
"type": "text"
}
]
},
"secret": {
"content": [
{
"text": "chut",
"type": "text"
}
]
},
"struct": {
"content": [],
"structuredContent": {
"a": "<x> & y",
"b": [
1,
2
]
}
},
"titled": {
"content": [
{
"text": "ok",
"type": "text"
}
]
}
},
"tools/list": {
"tools": [
{
"description": "Long",
"inputSchema": {
"type": "object"
},
"name": "big"
},
{
"description": "Nom pointé",
"inputSchema": {
"properties": {},
"type": "object"
},
"name": "dot.tool"
},
{
"description": "Répète le texte <tel quel>",
"inputSchema": {
"properties": {
"text": {
"type": "string"
}
},
"type": "object"
},
"name": "echo"
},
{
"description": "Rien",
"inputSchema": {
"type": "object"
},
"name": "empty"
},
{
"description": "Échoue",
"inputSchema": {
"type": "object"
},
"name": "fail"
},
{
"description": "Un lien",
"inputSchema": {
"type": "object"
},
"name": "link"
},
{
"description": "Image et son",
"inputSchema": {
"type": "object"
},
"name": "media"
},
{
"description": "Deux blocs",
"inputSchema": {
"type": "object"
},
"name": "multi"
},
{
"description": "Masqué",
"inputSchema": {
"type": "object"
},
"name": "secret"
},
{
"description": "Structuré",
"inputSchema": {
"type": "object"
},
"name": "struct"
},
{
"inputSchema": {
"type": "object"
},
"name": "titled",
"title": "Titre seul"
}
]
}
}
}
+1
View File
@@ -110,6 +110,7 @@ async fn facade(backend: SocketAddr) -> (SocketAddr, oneshot::Sender<()>) {
engine_ctl: None,
system: None,
projects: false,
mcp: None,
};
tokio::spawn(loki_rs::proxy::serve(l, cfg, async move {
let _ = rx.await;