Its tools say what llmgw ctl does: cluster_status, list_nodes, list_models, list_replicas, explain_placement, node_logs; none changes anything. The admin API serves it at POST /api/v1/mcp, and the gateway answers a server registered as llmgw:admin in its own process, so no token is kept and each control plane serves it. ctl mcp add takes --bearer-file for a token sent as Authorization: Bearer. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
4453f469e1
commit
4eb49d0e4c
11 files changed
+402
-7
No files matched your search
@@ -35,6 +35,14 @@ APIs change between minors without a deprecation period.
|
||||
server is told the calling key, person and project. Each call is bounded in time and size, and its progress passes
|
||||
on to a streaming client. Every call is in the audit log with its arguments' size and digest, never their text. The
|
||||
admin API has `/api/v1/mcp/servers`.
|
||||
- **The cluster as a read-only MCP server.** Registered as `llmgw ctl mcp add llmgw --url llmgw:admin` and granted
|
||||
as `llmgw.*`, its tools say what `llmgw ctl` does:
|
||||
- `cluster_status`, `list_nodes`, `list_models`, `list_replicas`, `explain_placement`, `node_logs`;
|
||||
- nothing that changes the cluster.
|
||||
|
||||
It is answered in the control plane's own process, so there is no token to keep and every control plane serves it.
|
||||
The admin API serves the same at `POST /api/v1/mcp`. `ctl mcp add --bearer-file` keeps a token, sent as
|
||||
`Authorization: Bearer`.
|
||||
|
||||
- **Large models across cards and nodes.** For a model of experts on more than one card, the plan now adds vLLM's
|
||||
`--enable-expert-parallel`. It reads the expert count from `config.json` under any of its names, and sizes a model
|
||||
|
||||
@@ -1939,6 +1939,51 @@
|
||||
}
|
||||
}
|
||||
},
|
||||
"/api/v1/mcp": {
|
||||
"post": {
|
||||
"operationId": "postMcp",
|
||||
"summary": "This cluster as a read-only MCP server: status, nodes, models, replicas, placement, logs",
|
||||
"description": "JSON-RPC: initialize, tools/list, tools/call. Registered with the gateway as llmgw:admin.",
|
||||
"tags": [
|
||||
"mcp"
|
||||
],
|
||||
"parameters": [
|
||||
{
|
||||
"$ref": "#/components/parameters/Actor"
|
||||
}
|
||||
],
|
||||
"requestBody": {
|
||||
"required": true,
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"type": "object",
|
||||
"additionalProperties": {}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "OK",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"type": "object",
|
||||
"additionalProperties": {}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"401": {
|
||||
"$ref": "#/components/responses/Unauthorized"
|
||||
},
|
||||
"default": {
|
||||
"$ref": "#/components/responses/Error"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"/api/v1/mcp/servers": {
|
||||
"get": {
|
||||
"operationId": "getMcpServers",
|
||||
|
||||
@@ -2537,6 +2537,24 @@ llmgw keys limits --scope team --id Acme/support --tools 'jira.*'
|
||||
every five minutes. `--disabled` keeps a server but serves none of its
|
||||
tools.
|
||||
|
||||
**This cluster's own, read only.** `llmgw ctl mcp add llmgw --url
|
||||
llmgw:admin` registers the cluster itself as a server: tools that say what
|
||||
`llmgw ctl` does, and change nothing. It is answered in the control plane's
|
||||
own process, with no address and no token to give.
|
||||
|
||||
| Tool | Says |
|
||||
|---|---|
|
||||
| `llmgw.cluster_status` | control planes, store, nodes, models against what they asked for, GPUs, what is unhealthy |
|
||||
| `llmgw.list_nodes` | each node's state, GPUs, load, fabrics, agent version and engines |
|
||||
| `llmgw.list_models` | each model's replicas ready against wanted, profiles, why one is parked |
|
||||
| `llmgw.list_replicas` | each replica's shards: node, GPUs, state (`model` for one model's) |
|
||||
| `llmgw.explain_placement` | why a model is not placed, node by node |
|
||||
| `llmgw.node_logs` | a node's last agent log lines, or an engine's (`source`) |
|
||||
|
||||
Grant it as any other: `llmgw keys limits --scope key --id KEY-ID --tools
|
||||
'llmgw.*'`. It is the same as an administrator's view, so grant it to the
|
||||
people who would have `ctl`.
|
||||
|
||||
Servers over HTTP only for now (the streamable HTTP transport). A server
|
||||
that runs as a local process (stdio) is planned.
|
||||
|
||||
|
||||
+15
-2
@@ -31,11 +31,13 @@ func newMCPCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
|
||||
func newMCPAddCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
|
||||
var req adminapi.MCPServerRequest
|
||||
var headers, refs, secretFiles []string
|
||||
var bearerFile string
|
||||
var disabled bool
|
||||
c := &cobra.Command{
|
||||
Use: "add NAME --url URL",
|
||||
Short: "Register an MCP server, or change one",
|
||||
Example: " llmgw ctl mcp add jira --url https://jira.internal/mcp --secret-file Authorization=/root/jira-token\n" +
|
||||
Example: " llmgw ctl mcp add jira --url https://jira.internal/mcp --bearer-file /root/jira-token\n" +
|
||||
" llmgw ctl mcp add llmgw --url llmgw:admin --description \"this cluster, read only\"\n" +
|
||||
" llmgw ctl mcp add wiki --url https://wiki.internal/mcp --header-ref Authorization=vault:secret/mcp/wiki#token\n" +
|
||||
" llmgw ctl mcp add jira --disabled # off, kept",
|
||||
Args: cobra.ExactArgs(1),
|
||||
@@ -63,6 +65,16 @@ func newMCPAddCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
|
||||
}
|
||||
req.Secrets[header] = strings.TrimSpace(string(b))
|
||||
}
|
||||
if bearerFile != "" {
|
||||
b, err := os.ReadFile(bearerFile) //nolint:gosec // a file the operator names
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if req.Secrets == nil {
|
||||
req.Secrets = map[string]string{}
|
||||
}
|
||||
req.Secrets["Authorization"] = "Bearer " + strings.TrimSpace(string(b))
|
||||
}
|
||||
if cmd.Flags().Changed("disabled") {
|
||||
on := !disabled
|
||||
req.Enabled = &on
|
||||
@@ -85,11 +97,12 @@ func newMCPAddCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
|
||||
},
|
||||
}
|
||||
f := c.Flags()
|
||||
f.StringVar(&req.URL, "url", "", "the server's streamable HTTP address, https://…/mcp")
|
||||
f.StringVar(&req.URL, "url", "", "the server's streamable HTTP address, https://…/mcp; llmgw:admin for this cluster, read only")
|
||||
f.StringVar(&req.Description, "description", "", "what it is, shown beside its tools")
|
||||
f.StringArrayVar(&headers, "header", nil, "NAME=VALUE sent as written, for what is not secret; repeatable")
|
||||
f.StringArrayVar(&refs, "header-ref", nil, "NAME=REF read at each call: a file on every control plane, env:, vault: or store:; repeatable")
|
||||
f.StringArrayVar(&secretFiles, "secret-file", nil, "NAME=FILE read now and kept in the store, sealed: a token; repeatable")
|
||||
f.StringVar(&bearerFile, "bearer-file", "", "a token FILE, read now and kept sealed, sent as Authorization: Bearer")
|
||||
f.BoolVar(&disabled, "disabled", false, "keep it but serve none of its tools")
|
||||
f.IntVar(&req.TimeoutSeconds, "timeout", 0, "seconds one call may take (default 60)")
|
||||
return c
|
||||
|
||||
@@ -440,6 +440,7 @@ func newServerCmd() *cobra.Command {
|
||||
// The cluster's settings, each change checked against this
|
||||
// machine's own keys before it is kept.
|
||||
api.SetMCP(st)
|
||||
rt.SetAdminMCP(http.HandlerFunc(api.MCP))
|
||||
api.SetAuth(&adminapi.AuthAdmin{Status: rt.AuthStatus, TestLDAP: rt.TestDirectory, Secrets: st, Log: log})
|
||||
api.SetSettings(&adminapi.Settings{Store: st, LocalKeys: config.LocalKeys(),
|
||||
Check: func(body []byte) error { return checkClusterSettings(configPath, body) }})
|
||||
|
||||
@@ -76,6 +76,7 @@ func (a *API) registerMCP(mux Mux) {
|
||||
}
|
||||
return true
|
||||
}
|
||||
mux.HandleFunc("POST /api/v1/mcp", a.MCP)
|
||||
mux.HandleFunc("GET /api/v1/mcp/servers", func(w http.ResponseWriter, r *http.Request) {
|
||||
if !need(w) {
|
||||
return
|
||||
@@ -138,7 +139,7 @@ func (a *API) registerMCP(mux Mux) {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, testMCPServer(r.Context(), m))
|
||||
writeJSON(w, http.StatusOK, a.testMCPServer(r.Context(), m))
|
||||
})
|
||||
}
|
||||
|
||||
@@ -159,8 +160,8 @@ func (a *API) mcpServerFrom(ctx context.Context, name string, req MCPServerReque
|
||||
if req.URL != "" {
|
||||
m.URL = req.URL
|
||||
}
|
||||
if u, err := url.Parse(m.URL); err != nil || (u.Scheme != "https" && u.Scheme != "http") || u.Host == "" {
|
||||
return m, errors.New("the server's url is an http:// or https:// address")
|
||||
if u, err := url.Parse(m.URL); m.URL != AdminMCPAddress && (err != nil || (u.Scheme != "https" && u.Scheme != "http") || u.Host == "") {
|
||||
return m, errors.New("the server's url is an http:// or https:// address, or " + AdminMCPAddress + " for this cluster's own")
|
||||
}
|
||||
if req.Description != "" {
|
||||
m.Description = req.Description
|
||||
@@ -196,7 +197,7 @@ func (a *API) mcpServerFrom(ctx context.Context, name string, req MCPServerReque
|
||||
|
||||
// testMCPServer asks a server for its tools, with its credentials, as /mcp
|
||||
// would.
|
||||
func testMCPServer(ctx context.Context, m store.MCPServer) MCPTestView {
|
||||
func (a *API) testMCPServer(ctx context.Context, m store.MCPServer) MCPTestView {
|
||||
ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
|
||||
defer cancel()
|
||||
c := &mcpclient.Client{URL: m.URL, Header: func(ctx context.Context) (http.Header, error) {
|
||||
@@ -213,6 +214,9 @@ func testMCPServer(ctx context.Context, m store.MCPServer) MCPTestView {
|
||||
}
|
||||
return h, nil
|
||||
}}
|
||||
if m.URL == AdminMCPAddress {
|
||||
c.URL, c.HTTP = InProcessURL, &http.Client{Transport: InProcess(http.HandlerFunc(a.MCP))}
|
||||
}
|
||||
tools, err := c.Tools(ctx)
|
||||
if err != nil {
|
||||
return MCPTestView{Error: err.Error()}
|
||||
|
||||
@@ -0,0 +1,186 @@
|
||||
package adminapi
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
// The cluster as an MCP server, read only: what `llmgw ctl status`, `nodes`,
|
||||
// `models`, `replicas`, `explain-placement` and `logs` show, as tools an
|
||||
// agent can be given. It answers at POST /api/v1/mcp, behind the admin
|
||||
// token, and in-process to the gateway as the MCP server llmgw:admin, which
|
||||
// an administrator registers like any other and grants like any other.
|
||||
// Nothing here changes the cluster.
|
||||
|
||||
// AdminMCPAddress is the address that registers this server with the
|
||||
// gateway: called in-process, with no token and no network.
|
||||
const AdminMCPAddress = "llmgw:admin"
|
||||
|
||||
// adminMCPAnswerBytes bounds one tool's answer: it becomes a prompt.
|
||||
const adminMCPAnswerBytes = 256 << 10
|
||||
|
||||
var adminMCPTools = []map[string]any{
|
||||
adminTool("cluster_status", "The whole cluster in one answer: control planes, store, nodes ready, models with the replicas they "+
|
||||
"asked for, GPUs in use, and what is unhealthy or idle.", nil, nil),
|
||||
adminTool("list_nodes", "Every node: its state, GPUs and their use, CPU and memory load, fabrics, agent version and engines.", nil, nil),
|
||||
adminTool("list_models", "Every model in the catalog: its replicas ready against wanted, state, profiles and why one is parked.", nil, nil),
|
||||
adminTool("list_replicas", "Every replica, with its shards: the node and GPUs each is on, its state and how long it has run.",
|
||||
map[string]any{"model": map[string]any{"type": "string", "description": "Only this model's replicas."}}, nil),
|
||||
adminTool("explain_placement", "Why a model is not placed: each node, and what kept the model off it.",
|
||||
map[string]any{"model": map[string]any{"type": "string", "description": "The model's name or an alias."}}, []string{"model"}),
|
||||
adminTool("node_logs", "The last lines of a node's agent log, or of one of its engines.",
|
||||
map[string]any{
|
||||
"node": map[string]any{"type": "string", "description": "The node's id."},
|
||||
"lines": map[string]any{"type": "integer", "description": "How many lines, up to 300; 100 when absent."},
|
||||
"source": map[string]any{"type": "string", "description": "agent when absent, or a shard id for that engine's output."},
|
||||
}, []string{"node"}),
|
||||
}
|
||||
|
||||
func adminTool(name, desc string, props map[string]any, required []string) map[string]any {
|
||||
schema := map[string]any{"type": "object", "properties": map[string]any{}}
|
||||
if props != nil {
|
||||
schema["properties"] = props
|
||||
}
|
||||
if len(required) > 0 {
|
||||
schema["required"] = required
|
||||
}
|
||||
return map[string]any{"name": name, "description": desc, "inputSchema": schema,
|
||||
"annotations": map[string]any{"readOnlyHint": true, "openWorldHint": false}}
|
||||
}
|
||||
|
||||
// MCP serves the cluster as a read-only MCP server: JSON-RPC over POST, the
|
||||
// streamable HTTP transport answered as plain JSON, no session.
|
||||
func (a *API) MCP(w http.ResponseWriter, r *http.Request) {
|
||||
var msg struct {
|
||||
ID json.RawMessage `json:"id"`
|
||||
Method string `json:"method"`
|
||||
Params json.RawMessage `json:"params"`
|
||||
}
|
||||
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&msg); err != nil || msg.Method == "" {
|
||||
writeJSON(w, http.StatusBadRequest, map[string]any{"jsonrpc": "2.0", "id": nil,
|
||||
"error": map[string]any{"code": -32600, "message": "expected one JSON-RPC request object"}})
|
||||
return
|
||||
}
|
||||
if len(msg.ID) == 0 {
|
||||
w.WriteHeader(http.StatusAccepted)
|
||||
return
|
||||
}
|
||||
answer := func(result any) {
|
||||
writeJSON(w, http.StatusOK, map[string]any{"jsonrpc": "2.0", "id": msg.ID, "result": result})
|
||||
}
|
||||
switch msg.Method {
|
||||
case "initialize":
|
||||
var p struct {
|
||||
ProtocolVersion string `json:"protocolVersion"`
|
||||
}
|
||||
_ = json.Unmarshal(msg.Params, &p)
|
||||
version := p.ProtocolVersion
|
||||
if version != "2025-03-26" && version != "2025-06-18" {
|
||||
version = "2025-06-18"
|
||||
}
|
||||
answer(map[string]any{"protocolVersion": version, "capabilities": map[string]any{"tools": map[string]any{}},
|
||||
"serverInfo": map[string]any{"name": "llmgw-admin", "version": "1"},
|
||||
"instructions": "This llmgw cluster, read only: its control planes, nodes, models and replicas, why a model is " +
|
||||
"not placed, and a node's logs. Start with cluster_status."})
|
||||
case "ping":
|
||||
answer(map[string]any{})
|
||||
case "tools/list":
|
||||
answer(map[string]any{"tools": adminMCPTools})
|
||||
case "tools/call":
|
||||
var call struct {
|
||||
Name string `json:"name"`
|
||||
Arguments struct {
|
||||
Model string `json:"model"`
|
||||
Node string `json:"node"`
|
||||
Source string `json:"source"`
|
||||
Lines int `json:"lines"`
|
||||
} `json:"arguments"`
|
||||
}
|
||||
_ = json.Unmarshal(msg.Params, &call)
|
||||
answer(a.callAdminTool(r, call.Name, call.Arguments.Model, call.Arguments.Node, call.Arguments.Source, call.Arguments.Lines))
|
||||
default:
|
||||
writeJSON(w, http.StatusOK, map[string]any{"jsonrpc": "2.0", "id": msg.ID,
|
||||
"error": map[string]any{"code": -32601, "message": "method not found: " + msg.Method}})
|
||||
}
|
||||
}
|
||||
|
||||
// callAdminTool answers a tool from the API's own handler for it, so the
|
||||
// tool says what `llmgw ctl` would.
|
||||
func (a *API) callAdminTool(r *http.Request, name, model, node, source string, lines int) map[string]any {
|
||||
var h http.HandlerFunc
|
||||
q := url.Values{}
|
||||
path := ""
|
||||
switch name {
|
||||
case "cluster_status":
|
||||
h, path = a.summary, "/api/v1/summary"
|
||||
case "list_nodes":
|
||||
h, path = a.listNodes, "/api/v1/nodes"
|
||||
case "list_models":
|
||||
h, path = a.listModels, "/api/v1/models"
|
||||
case "list_replicas":
|
||||
h, path = a.listReplicas, "/api/v1/replicas"
|
||||
if model != "" {
|
||||
q.Set("model", model)
|
||||
}
|
||||
case "explain_placement":
|
||||
if model == "" {
|
||||
return adminToolText("explain_placement needs the model", true)
|
||||
}
|
||||
h, path = a.explainPlacement, "/api/v1/explain-placement"
|
||||
q.Set("model", model)
|
||||
case "node_logs":
|
||||
if node == "" {
|
||||
return adminToolText("node_logs needs the node", true)
|
||||
}
|
||||
if lines <= 0 {
|
||||
lines = 100
|
||||
}
|
||||
h, path = a.nodeLogs, "/api/v1/nodes/"+url.PathEscape(node)+"/logs"
|
||||
q.Set("lines", strconv.Itoa(min(lines, 300)))
|
||||
if source != "" {
|
||||
q.Set("source", source)
|
||||
}
|
||||
default:
|
||||
return adminToolText("no tool "+name, true)
|
||||
}
|
||||
req := httptest.NewRequestWithContext(r.Context(), http.MethodGet, path+"?"+q.Encode(), nil)
|
||||
if name == "node_logs" {
|
||||
req.SetPathValue("id", node)
|
||||
}
|
||||
rec := httptest.NewRecorder()
|
||||
h(rec, req)
|
||||
body := bytes.TrimSpace(rec.Body.Bytes())
|
||||
if rec.Code >= 300 {
|
||||
return adminToolText(fmt.Sprintf("%s: %s", name, body), true)
|
||||
}
|
||||
if len(body) > adminMCPAnswerBytes {
|
||||
body = append(body[:adminMCPAnswerBytes], []byte("… (cut: the answer is larger than an agent should read)")...)
|
||||
}
|
||||
return adminToolText(string(body), false)
|
||||
}
|
||||
|
||||
func adminToolText(text string, isError bool) map[string]any {
|
||||
return map[string]any{"content": []any{map[string]any{"type": "text", "text": text}}, "isError": isError}
|
||||
}
|
||||
|
||||
// InProcessURL is the address an in-process client posts to; the transport
|
||||
// ignores it.
|
||||
const InProcessURL = "http://llmgw-admin/api/v1/mcp"
|
||||
|
||||
// InProcess is a transport that answers every request from h, in this
|
||||
// process: the gateway's way to the cluster's own MCP server.
|
||||
func InProcess(h http.Handler) http.RoundTripper { return inProcess{h} }
|
||||
|
||||
type inProcess struct{ h http.Handler }
|
||||
|
||||
func (t inProcess) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||
rec := httptest.NewRecorder()
|
||||
t.h.ServeHTTP(rec, req)
|
||||
return rec.Result(), nil
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package adminapi
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func adminMCP(t *testing.T, a *API, method string, params any) map[string]any {
|
||||
t.Helper()
|
||||
b, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": 1, "method": method, "params": params})
|
||||
rec := httptest.NewRecorder()
|
||||
a.MCP(rec, httptest.NewRequest(http.MethodPost, "/api/v1/mcp", bytes.NewReader(b)))
|
||||
var resp struct {
|
||||
Result map[string]any `json:"result"`
|
||||
}
|
||||
if rec.Code != 200 || json.Unmarshal(rec.Body.Bytes(), &resp) != nil {
|
||||
t.Fatalf("%s: %d %s", method, rec.Code, rec.Body.String())
|
||||
}
|
||||
return resp.Result
|
||||
}
|
||||
|
||||
func TestTheClusterIsAReadOnlyMCPServer(t *testing.T) {
|
||||
a, _ := setup(t, testModels())
|
||||
if init := adminMCP(t, a, "initialize", map[string]any{"protocolVersion": "2025-06-18"}); init["protocolVersion"] != "2025-06-18" {
|
||||
t.Errorf("initialize %v", init)
|
||||
}
|
||||
tools := adminMCP(t, a, "tools/list", map[string]any{})["tools"].([]any)
|
||||
var names []string
|
||||
for _, tl := range tools {
|
||||
m := tl.(map[string]any)
|
||||
names = append(names, m["name"].(string))
|
||||
if m["annotations"].(map[string]any)["readOnlyHint"] != true {
|
||||
t.Errorf("%s is not marked read only", m["name"])
|
||||
}
|
||||
}
|
||||
if got := strings.Join(names, ","); got != "cluster_status,list_nodes,list_models,list_replicas,explain_placement,node_logs" {
|
||||
t.Errorf("tools %s", got)
|
||||
}
|
||||
// Each answers what the API would.
|
||||
for _, name := range []string{"cluster_status", "list_nodes", "list_models", "list_replicas"} {
|
||||
res := adminMCP(t, a, "tools/call", map[string]any{"name": name, "arguments": map[string]any{}})
|
||||
text := res["content"].([]any)[0].(map[string]any)["text"].(string)
|
||||
if res["isError"] == true || !json.Valid([]byte(text)) {
|
||||
t.Errorf("%s: %v", name, res)
|
||||
}
|
||||
}
|
||||
if res := adminMCP(t, a, "tools/call", map[string]any{"name": "explain_placement", "arguments": map[string]any{}}); res["isError"] != true {
|
||||
t.Errorf("explain_placement without a model: %v", res)
|
||||
}
|
||||
if res := adminMCP(t, a, "tools/call", map[string]any{"name": "remove_model", "arguments": map[string]any{}}); res["isError"] != true {
|
||||
t.Errorf("a tool that would change something: %v", res)
|
||||
}
|
||||
}
|
||||
|
||||
// The gateway reaches it in this process, as llmgw:admin.
|
||||
func TestTheClustersOwnMCPServerIsReachedInProcess(t *testing.T) {
|
||||
a, _ := setup(t, testModels())
|
||||
c := &http.Client{Transport: InProcess(http.HandlerFunc(a.MCP))}
|
||||
b, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
|
||||
resp, err := c.Post(InProcessURL, "application/json", bytes.NewReader(b))
|
||||
if err != nil || resp.StatusCode != 200 {
|
||||
t.Fatalf("%v %v", resp, err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
var out struct {
|
||||
Result struct {
|
||||
Tools []any `json:"tools"`
|
||||
} `json:"result"`
|
||||
}
|
||||
if json.NewDecoder(resp.Body).Decode(&out) != nil || len(out.Result.Tools) != 6 {
|
||||
t.Errorf("tools %+v", out)
|
||||
}
|
||||
}
|
||||
@@ -408,6 +408,9 @@ var Routes = []Route{
|
||||
Request: SecretRequest{}, Response: map[string]string{}},
|
||||
|
||||
// The company's own MCP servers, served through /mcp.
|
||||
{Method: "POST", Path: "/api/v1/mcp", Summary: "This cluster as a read-only MCP server: status, nodes, models, replicas, placement, logs",
|
||||
Description: "JSON-RPC: initialize, tools/list, tools/call. Registered with the gateway as llmgw:admin.",
|
||||
Request: map[string]any{}, Response: map[string]any{}},
|
||||
{Method: "GET", Path: "/api/v1/mcp/servers", Summary: "The company's MCP servers /mcp serves, and how each is reached",
|
||||
Response: []MCPServerView{}},
|
||||
{Method: "PUT", Path: "/api/v1/mcp/servers/{name}", Summary: "Add an MCP server, or change one: its address, headers and the secrets they carry",
|
||||
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/auth"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/mcpclient"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/secrets"
|
||||
storepkg "gitea.mynodes.xyz/ilker/llmgw/internal/store"
|
||||
@@ -61,6 +62,21 @@ type mcpGateway struct {
|
||||
mu sync.Mutex
|
||||
readAt time.Time
|
||||
upstream map[string]*mcpUpstream
|
||||
// local is this cluster's own MCP server, the admin API's, which a
|
||||
// server registered as llmgw:admin is answered by in this process.
|
||||
local http.Handler
|
||||
}
|
||||
|
||||
// SetAdminMCP is the cluster's own read-only MCP server, served to keys
|
||||
// granted it once an administrator registers it as llmgw:admin.
|
||||
func (r *Router) SetAdminMCP(h http.Handler) {
|
||||
if r.mcpgw == nil {
|
||||
return
|
||||
}
|
||||
r.mcpgw.mu.Lock()
|
||||
defer r.mcpgw.mu.Unlock()
|
||||
r.mcpgw.local = h
|
||||
r.mcpgw.readAt = time.Time{}
|
||||
}
|
||||
|
||||
// mcpUpstream is one server: its settings, its client, its tools.
|
||||
@@ -133,8 +149,12 @@ func sameMCPServer(a, b storepkg.MCPServer) bool {
|
||||
|
||||
func (g *mcpGateway) upstreamFor(s storepkg.MCPServer) *mcpUpstream {
|
||||
cfg := s
|
||||
url, client := s.URL, g.http
|
||||
if s.URL == adminapi.AdminMCPAddress && g.local != nil {
|
||||
url, client = adminapi.InProcessURL, &http.Client{Transport: adminapi.InProcess(g.local)}
|
||||
}
|
||||
return &mcpUpstream{cfg: s, client: &mcpclient.Client{
|
||||
URL: s.URL, HTTP: g.http, MaxBytes: mcpAnswerBytes,
|
||||
URL: url, HTTP: client, MaxBytes: mcpAnswerBytes,
|
||||
// Read at each request, so a rotated credential is used at once.
|
||||
Header: func(ctx context.Context) (http.Header, error) {
|
||||
h := http.Header{}
|
||||
|
||||
@@ -110,3 +110,23 @@ func toolResultText(res map[string]any) string {
|
||||
text, _ := content[0].(map[string]any)["text"].(string)
|
||||
return text
|
||||
}
|
||||
|
||||
// A server registered as llmgw:admin is answered in this process by the
|
||||
// handler the control plane gave, with no address to reach.
|
||||
func TestTheClustersOwnMCPServerIsServedInProcess(t *testing.T) {
|
||||
srv := mcptest.New()
|
||||
defer srv.Close()
|
||||
rt, granted, _ := companyRouter(t, srv)
|
||||
rt.SetAdminMCP(srv.Config.Handler)
|
||||
if err := rt.store.PutMCPServer(t.Context(), store.MCPServer{Name: "llmgw", URL: "llmgw:admin", Enabled: true}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := rt.store.PutLimits(t.Context(), store.Limits{Scope: store.ScopeKey, ScopeID: "key-g", AllowedTools: []string{"jira.echo", "llmgw.*"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rt.mcpgw.readAt = time.Time{}
|
||||
res, _ := mcp(t, rt.Handler(), granted, "tools/call", map[string]any{"name": "llmgw.echo", "arguments": map[string]any{"text": "inside"}}, false)
|
||||
if !strings.Contains(toolResultText(res), "inside") {
|
||||
t.Errorf("result %v", res)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user