The Go SDK serves as the module the runtime names, answers calls concurrently, and emits through the runtime (hq ADR 0193)
What the TypeScript SDK's stdio loop does, in Go: an empty name serves as MESH_SERVED_MODULE; each call runs beside the loop so a tool waiting on the runtime does not deadlock it; Emit asks the runtime (mesh/publish) with the same envelope and headers. README says how to install it and how releases are tagged.
This commit is contained in:
jochen
+13
-3
@@ -1,8 +1,18 @@
|
||||
# mesh-sdk for Go
|
||||
|
||||
The skeleton (novox/hq ADR 0188): `stdio.Serve(name, tools)` speaks MCP over stdio to the node's
|
||||
tool runtime. A `Tool` is a name, a description, an argument schema and a function. A tool named
|
||||
`<seat>.<verb>` is the seat's implementation. Standard library only.
|
||||
`stdio.Serve("", tools)` speaks MCP over stdio to the node's tool runtime, as the module the runtime
|
||||
names (novox/hq ADR 0188, ADR 0193). A `Tool` is a name, a description, an argument schema and a
|
||||
function; calls run concurrently. A tool named `<seat>.<verb>` is the seat's implementation.
|
||||
`stdio.Emit(type, body)` publishes an event as the module, through the runtime. Standard library only.
|
||||
|
||||
## Installing
|
||||
|
||||
```
|
||||
GOPRIVATE=git.novox.be go get git.novox.be/novox/mesh-sdk/go@latest
|
||||
```
|
||||
|
||||
Releases are tagged `go/vX.Y.Z` in this repository, the subdirectory form Go's module system expects,
|
||||
at the same version as the TypeScript package.
|
||||
|
||||
What it holds, under ADR 0039's test: the stdio loop and the tool type. Nothing else — no bus, no
|
||||
client, nothing of any module's. `go build ./example` and pipe a `tools/list` request into it.
|
||||
|
||||
+170
-30
@@ -1,4 +1,4 @@
|
||||
// Package stdio is the Novox Mesh SDK for Go — the skeleton (novox/hq ADR 0188).
|
||||
// Package stdio is the Novox Mesh SDK for Go (novox/hq ADR 0188, ADR 0193).
|
||||
//
|
||||
// A tools bundle is a process the node's runtime launches and speaks MCP over stdio to:
|
||||
// `initialize`, `tools/list` once, `tools/call` per call, newline-framed JSON-RPC on stdin and
|
||||
@@ -7,20 +7,31 @@
|
||||
// module's own.
|
||||
//
|
||||
// No transport here: the bus, the subjects and the memberships are the runtime's (ADR 0039, 0175).
|
||||
// stdout is the protocol; log to stderr.
|
||||
// What a bundle asks of the bus — emitting an event — it asks of the runtime over the same channel,
|
||||
// and the runtime does it as the module (ADR 0193). stdout is the protocol; log to stderr.
|
||||
package stdio
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Protocol is the MCP version this speaks; the runtime says the same.
|
||||
const Protocol = "2025-03-26"
|
||||
|
||||
// ServedAs is the word the runtime sets for a bundle it launches: the module it serves it as
|
||||
// (ADR 0193). A bundle served under that name lists its own tools by their names.
|
||||
const ServedAs = "MESH_SERVED_MODULE"
|
||||
|
||||
// Tool is one tool a bundle serves.
|
||||
type Tool struct {
|
||||
Name string
|
||||
@@ -30,48 +41,116 @@ type Tool struct {
|
||||
Run func(args map[string]any) (any, error)
|
||||
}
|
||||
|
||||
type request struct {
|
||||
ID *json.RawMessage `json:"id"`
|
||||
Method string `json:"method"`
|
||||
Params struct {
|
||||
Name string `json:"name"`
|
||||
Arguments map[string]any `json:"arguments"`
|
||||
} `json:"params"`
|
||||
type message struct {
|
||||
ID *json.RawMessage `json:"id,omitempty"`
|
||||
Method string `json:"method,omitempty"`
|
||||
Params json.RawMessage `json:"params,omitempty"`
|
||||
Result json.RawMessage `json:"result,omitempty"`
|
||||
Error *struct {
|
||||
Message string `json:"message"`
|
||||
} `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// Serve answers the runtime on stdin and stdout until stdin closes.
|
||||
// channel is the one conversation with the runtime: what this bundle writes, serialised, and what
|
||||
// it asked and waits for.
|
||||
type channel struct {
|
||||
mu sync.Mutex
|
||||
enc *json.Encoder
|
||||
next int
|
||||
waiting map[string]chan message
|
||||
}
|
||||
|
||||
func (c *channel) say(m map[string]any) {
|
||||
m["jsonrpc"] = "2.0"
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
_ = c.enc.Encode(m)
|
||||
}
|
||||
|
||||
// ask sends the runtime a request and waits for its answer.
|
||||
func (c *channel) ask(method string, params any, timeout time.Duration) (json.RawMessage, error) {
|
||||
c.mu.Lock()
|
||||
c.next++
|
||||
id := "bundle-" + strconv.Itoa(c.next)
|
||||
answer := make(chan message, 1)
|
||||
c.waiting[id] = answer
|
||||
c.mu.Unlock()
|
||||
c.say(map[string]any{"id": id, "method": method, "params": params})
|
||||
select {
|
||||
case m := <-answer:
|
||||
if m.Error != nil {
|
||||
return nil, errors.New(m.Error.Message)
|
||||
}
|
||||
return m.Result, nil
|
||||
case <-time.After(timeout):
|
||||
c.mu.Lock()
|
||||
delete(c.waiting, id)
|
||||
c.mu.Unlock()
|
||||
return nil, fmt.Errorf("the runtime did not answer %s in %s", method, timeout)
|
||||
}
|
||||
}
|
||||
|
||||
var (
|
||||
current *channel
|
||||
currentMu sync.Mutex
|
||||
)
|
||||
|
||||
// Serve answers the runtime on stdin and stdout until stdin closes. An empty name serves as the
|
||||
// module the runtime named in MESH_SERVED_MODULE.
|
||||
func Serve(name string, tools []Tool) error {
|
||||
return ServeOn(name, tools, os.Stdin, os.Stdout)
|
||||
}
|
||||
|
||||
// ServeOn is Serve over any pair of streams, for a test.
|
||||
func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error {
|
||||
if name == "" {
|
||||
name = os.Getenv(ServedAs)
|
||||
}
|
||||
if name == "" {
|
||||
name = "bundle"
|
||||
}
|
||||
byName := map[string]Tool{}
|
||||
for _, t := range tools {
|
||||
byName[t.Name] = t
|
||||
}
|
||||
enc := json.NewEncoder(out)
|
||||
say := func(m map[string]any) { m["jsonrpc"] = "2.0"; _ = enc.Encode(m) }
|
||||
c := &channel{enc: json.NewEncoder(out), waiting: map[string]chan message{}}
|
||||
currentMu.Lock()
|
||||
current = c
|
||||
currentMu.Unlock()
|
||||
scanner := bufio.NewScanner(in)
|
||||
scanner.Buffer(make([]byte, 1<<20), 1<<24)
|
||||
for scanner.Scan() {
|
||||
var req request
|
||||
if err := json.Unmarshal(scanner.Bytes(), &req); err != nil {
|
||||
var m message
|
||||
if err := json.Unmarshal(scanner.Bytes(), &m); err != nil {
|
||||
continue
|
||||
}
|
||||
// The runtime's answer to something this bundle asked.
|
||||
if m.Method == "" && m.ID != nil {
|
||||
var key string
|
||||
if json.Unmarshal(*m.ID, &key) == nil {
|
||||
c.mu.Lock()
|
||||
answer, ok := c.waiting[key]
|
||||
delete(c.waiting, key)
|
||||
c.mu.Unlock()
|
||||
if ok {
|
||||
answer <- m
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
id := any(nil)
|
||||
if req.ID != nil {
|
||||
id = *req.ID
|
||||
if m.ID != nil {
|
||||
id = *m.ID
|
||||
}
|
||||
switch req.Method {
|
||||
switch m.Method {
|
||||
case "initialize":
|
||||
say(map[string]any{"id": id, "result": map[string]any{"protocolVersion": Protocol,
|
||||
c.say(map[string]any{"id": id, "result": map[string]any{"protocolVersion": Protocol,
|
||||
"capabilities": map[string]any{"tools": map[string]any{}},
|
||||
"serverInfo": map[string]any{"name": name, "version": "1"}}})
|
||||
case "notifications/initialized":
|
||||
case "ping":
|
||||
if id != nil {
|
||||
say(map[string]any{"id": id, "result": map[string]any{}})
|
||||
c.say(map[string]any{"id": id, "result": map[string]any{}})
|
||||
}
|
||||
case "tools/list":
|
||||
listed := make([]map[string]any, 0, len(tools))
|
||||
@@ -84,25 +163,86 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error {
|
||||
}
|
||||
listed = append(listed, map[string]any{"name": t.Name, "description": t.Description, "inputSchema": schema})
|
||||
}
|
||||
say(map[string]any{"id": id, "result": map[string]any{"tools": listed}})
|
||||
c.say(map[string]any{"id": id, "result": map[string]any{"tools": listed}})
|
||||
case "tools/call":
|
||||
t, ok := byName[req.Params.Name]
|
||||
var p struct {
|
||||
Name string `json:"name"`
|
||||
Arguments map[string]any `json:"arguments"`
|
||||
}
|
||||
_ = json.Unmarshal(m.Params, &p)
|
||||
t, ok := byName[p.Name]
|
||||
if !ok {
|
||||
say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": fmt.Sprintf("%s has no tool %s", name, req.Params.Name)}})
|
||||
c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": fmt.Sprintf("%s has no tool %s", name, p.Name)}})
|
||||
continue
|
||||
}
|
||||
result, err := t.Run(req.Params.Arguments)
|
||||
if err != nil {
|
||||
say(map[string]any{"id": id, "result": map[string]any{"content": []map[string]any{{"type": "text", "text": err.Error()}}, "isError": true}})
|
||||
continue
|
||||
}
|
||||
text, _ := json.Marshal(result)
|
||||
say(map[string]any{"id": id, "result": map[string]any{"content": []map[string]any{{"type": "text", "text": string(text)}}}})
|
||||
// Beside the loop, never in it: a tool that emits waits for the runtime's answer, which
|
||||
// arrives on the line this loop reads next — and calls do not queue behind a slow one.
|
||||
go func(id any, t Tool, args map[string]any) {
|
||||
if args == nil {
|
||||
args = map[string]any{}
|
||||
}
|
||||
result, err := t.Run(args)
|
||||
if err != nil {
|
||||
c.say(map[string]any{"id": id, "result": map[string]any{"content": []map[string]any{{"type": "text", "text": err.Error()}}, "isError": true}})
|
||||
return
|
||||
}
|
||||
text, _ := json.Marshal(result)
|
||||
c.say(map[string]any{"id": id, "result": map[string]any{"content": []map[string]any{{"type": "text", "text": string(text)}}}})
|
||||
}(id, t, p.Arguments)
|
||||
default:
|
||||
if id != nil {
|
||||
say(map[string]any{"id": id, "error": map[string]any{"code": -32601, "message": "no " + req.Method}})
|
||||
c.say(map[string]any{"id": id, "error": map[string]any{"code": -32601, "message": "no " + m.Method}})
|
||||
}
|
||||
}
|
||||
}
|
||||
return scanner.Err()
|
||||
}
|
||||
|
||||
// EmitOptions is the provenance an emitter may attach (ADR 0042).
|
||||
type EmitOptions struct {
|
||||
CausationID string
|
||||
Schema string
|
||||
}
|
||||
|
||||
// Emit asks the runtime to publish an event as this bundle's module, and returns once the bus has
|
||||
// accepted it (ADR 0193). The envelope and headers are the TypeScript SDK's exactly: the source and
|
||||
// node come from MESH_MODULE and MESH_NODE, which the runtime sets.
|
||||
func Emit(eventType string, body any, opts ...EmitOptions) error {
|
||||
currentMu.Lock()
|
||||
c := current
|
||||
currentMu.Unlock()
|
||||
if c == nil {
|
||||
return errors.New("emit before Serve: a bundle emits through the runtime that launched it")
|
||||
}
|
||||
source, node := os.Getenv("MESH_MODULE"), os.Getenv("MESH_NODE")
|
||||
if source == "" {
|
||||
source = "unknown"
|
||||
}
|
||||
if node == "" {
|
||||
node = "unknown"
|
||||
}
|
||||
headers := map[string]string{
|
||||
"x-event-id": eventID(), "x-source": source, "x-node": node,
|
||||
"x-time": time.Now().UTC().Format(time.RFC3339Nano), "content-type": "application/json",
|
||||
}
|
||||
for _, o := range opts {
|
||||
if o.CausationID != "" {
|
||||
headers["x-causation-id"] = o.CausationID
|
||||
}
|
||||
if o.Schema != "" {
|
||||
headers["x-schema"] = o.Schema
|
||||
}
|
||||
}
|
||||
_, err := c.ask("mesh/publish", map[string]any{"key": eventType, "node": node, "body": body, "headers": headers}, 30*time.Second)
|
||||
return err
|
||||
}
|
||||
|
||||
// eventID is a random UUID, version 4, as the TypeScript SDK's randomUUID.
|
||||
func eventID() string {
|
||||
var b [16]byte
|
||||
_, _ = rand.Read(b[:])
|
||||
b[6] = (b[6] & 0x0f) | 0x40
|
||||
b[8] = (b[8] & 0x3f) | 0x80
|
||||
h := hex.EncodeToString(b[:])
|
||||
return h[0:8] + "-" + h[8:12] + "-" + h[12:16] + "-" + h[16:20] + "-" + h[20:32]
|
||||
}
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
package stdio
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A Go bundle serves as the module the runtime names, answers calls concurrently, and emits through
|
||||
// the runtime with the TypeScript SDK's envelope (novox/hq ADR 0193).
|
||||
func TestABundleEmitsThroughTheRuntimeAndServesAsItsModule(t *testing.T) {
|
||||
t.Setenv(ServedAs, "chime")
|
||||
t.Setenv("MESH_MODULE", "chime")
|
||||
t.Setenv("MESH_NODE", "anchor")
|
||||
inR, inW := io.Pipe()
|
||||
outR, outW := io.Pipe()
|
||||
go func() {
|
||||
_ = ServeOn("", []Tool{{Name: "ring", Description: "rings", Run: func(map[string]any) (any, error) {
|
||||
if err := Emit("rang", map[string]any{"times": 1}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"rang": true}, nil
|
||||
}}}, inR, outW)
|
||||
}()
|
||||
lines := bufio.NewScanner(outR)
|
||||
next := func() map[string]any {
|
||||
done := make(chan map[string]any, 1)
|
||||
go func() {
|
||||
if lines.Scan() {
|
||||
var m map[string]any
|
||||
_ = json.Unmarshal(lines.Bytes(), &m)
|
||||
done <- m
|
||||
}
|
||||
}()
|
||||
select {
|
||||
case m := <-done:
|
||||
return m
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("the bundle said nothing")
|
||||
return nil
|
||||
}
|
||||
}
|
||||
send := func(m map[string]any) { b, _ := json.Marshal(m); _, _ = inW.Write(append(b, '\n')) }
|
||||
|
||||
send(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": map[string]any{}})
|
||||
if got := next()["result"].(map[string]any)["serverInfo"].(map[string]any)["name"]; got != "chime" {
|
||||
t.Fatalf("served as %v, want the module the runtime named", got)
|
||||
}
|
||||
send(map[string]any{"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": map[string]any{"name": "ring", "arguments": map[string]any{}}})
|
||||
asked := next()
|
||||
if asked["method"] != "mesh/publish" {
|
||||
t.Fatalf("the tool's emit was not asked of the runtime: %v", asked)
|
||||
}
|
||||
p := asked["params"].(map[string]any)
|
||||
h := p["headers"].(map[string]any)
|
||||
if p["key"] != "rang" || p["node"] != "anchor" || h["x-source"] != "chime" || h["content-type"] != "application/json" || len(h["x-event-id"].(string)) != 36 {
|
||||
t.Errorf("the envelope is not the SDK's: %v", p)
|
||||
}
|
||||
send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{}})
|
||||
answered := next()
|
||||
text := answered["result"].(map[string]any)["content"].([]any)[0].(map[string]any)["text"]
|
||||
if answered["id"] != float64(2) || text != `{"rang":true}` {
|
||||
t.Errorf("the call answered %v", answered)
|
||||
}
|
||||
_ = inW.Close()
|
||||
}
|
||||
Reference in New Issue
Block a user