src/io.bend

ezmcp@0.1.0.0 · 444 lines · 29 definitions

1# ezmcp/src/io: the effects. Every decision is src/web.bend's, src/line.bend's,
2# src/server.bend's or src/client.bend's; this module reads, runs the tool
3# runner, and writes, in order (MCP-TRUST-1). It reaches ezhttp's socket
4# effects, so no law file imports it.
5import Base
6import emerging-ezjson@1.2.0.0/main.bend as Json
7import emerging-ezjson@1.2.0.0/src/value.bend as Value
8import emerging-ezhttp@0.21.0.0/main.bend as Ez
9import emerging-ezhttp@0.21.0.0/src/http.bend as Http
10import emerging-ezhttp@0.21.0.0/src/client.bend as EzClient
11import emerging-ezhttp@0.21.0.0/src/server.bend as EzServer
12import ezmcp@0.1.0.0/src/rpc.bend as Rpc
13import ezmcp@0.1.0.0/src/server.bend as Server
14import ezmcp@0.1.0.0/src/web.bend as Web
15import ezmcp@0.1.0.0/src/line.bend as Line
16import ezmcp@0.1.0.0/src/client.bend as Client
17
18# --- running acts ---
19
20# one element of a batch, under revision `rev`: its reply, after running its
21# tool when it is a tool call; none for a notification or a response
22def act.reply(
23  ~C: Data,
24  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
25  +ctx: C,
26  +rev: String,
27  act: Server.Act
28) -> IO(Maybe<&2, Value.Json>):
29  match act:
30    case Server.Quiet{}:
31      IO.pure(Maybe<&2, Value.Json>, None{})
32    case Server.Send{reply}:
33      IO.pure(Maybe<&2, Value.Json>, Some{reply})
34    case Server.Fault{reply}:
35      IO.pure(Maybe<&2, Value.Json>, Some{reply})
36    case Server.Run{id, name, args}:
37      IO.bind(Rpc.Out, Maybe<&2, Value.Json>, run(ctx, name, args),
38        out => IO.pure(Maybe<&2, Value.Json>, Some{Server.ran(rev, id, out)}))
39    case Server.Batch{acts}:
40      IO.pure(Maybe<&2, Value.Json>, None{})
41
42# every element of a batch, in order
43def act.all(
44  ~C: Data,
45  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
46  +ctx: C,
47  +rev: String,
48  acts: List<&2, Server.Act>
49) -> IO(List<&2, Maybe<&2, Value.Json>>):
50  match acts:
51    case []:
52      IO.pure(List<&2, Maybe<&2, Value.Json>>, [])
53    case a <> rest:
54      IO.bind(Maybe<&2, Value.Json>, List<&2, Maybe<&2, Value.Json>>,
55        act.reply(C, run, ctx, rev, a),
56        r => IO.bind(List<&2, Maybe<&2, Value.Json>>,
57          List<&2, Maybe<&2, Value.Json>>, act.all(C, run, ctx, rev, rest),
58          rs => IO.pure(List<&2, Maybe<&2, Value.Json>>, r <> rs)))
59
60# --- streamable HTTP ---
61
62# act on the transport's decision: reply now, or run the tool and reply
63# with its result, or run a batch's acts and reply with their replies
64def http.web(
65  ~C: Data,
66  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
67  +ctx: C,
68  wid: Web.Web
69) -> IO(Http.Reply):
70  match wid:
71    case Web.Now{status, headers, body}:
72      IO.pure(Http.Reply, Http.Reply{status, headers, body})
73    case Web.Later{rev, id, name, args}:
74      IO.bind(Rpc.Out, Http.Reply, run(ctx, name, args),
75        out => IO.pure(Http.Reply, Web.reply(Web.ran(rev, id, out))))
76    case Web.Many{+rev, acts}:
77      IO.bind(List<&2, Maybe<&2, Value.Json>>, Http.Reply,
78        act.all(C, run, ctx, rev, acts),
79        rs => IO.pure(Http.Reply, Web.reply(Web.many(Server.batch.reply(rs)))))
80
81# one HTTP request to a server's endpoint at `path`
82def http.handle_with(
83  ~C: Data,
84  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
85  ~srv: Server.Server,
86  ~path: String,
87  ~origins: List<&2, String>,
88  +ctx: C,
89  req: Http.Request
90) -> IO(Http.Reply):
91  http.web(C, run, ctx, Web.decide(srv, path, origins, req))
92
93# --- serving HTTP ---
94#
95# ezhttp serves each connection on its own fork, reads and parses the
96# request, answers one it answers itself (a request that does not parse),
97# and calls the handler with the context for the rest (MCP-TRUST-2); the
98# handler below is the endpoint's
99
100# the endpoint's answer to one request ezhttp read, as its plain serves
101# take one: the request as text in, a text reply out
102def http.answer(
103  ~C: Data,
104  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
105  ~srv: Server.Server,
106  ~path: String,
107  ~origins: List<&2, String>,
108  +ctx: C,
109  req: EzServer.Req
110) -> IO(EzServer.Answer):
111  IO.bind(Http.Reply, EzServer.Answer,
112    http.handle_with(C, run, srv, path, origins, ctx, EzServer.server.plain(req)),
113    EzServer.server.text)
114
115# serve `limit` connections, at most `max` at once (0: no limit), then stop
116def http.serve_with(
117  ~C: Data,
118  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
119  ~srv: Server.Server,
120  ~path: String,
121  ~origins: List<&2, String>,
122  +ctx: C,
123  host: String,
124  port: U32,
125  limit: U32,
126  +max: U32
127) -> IO(Unit):
128  Ez.http.serve_io_with(~C,
129    ~(c => req => http.answer(C, run, srv, path, origins, c, req)), ctx, host,
130    port, limit, max)
131
132# serve until the process is stopped, at most `max` connections at once
133# (0: no limit)
134@unsafe def http.serve_forever_with(
135  ~C: Data,
136  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
137  ~srv: Server.Server,
138  ~path: String,
139  ~origins: List<&2, String>,
140  +ctx: C,
141  host: String,
142  port: U32,
143  +max: U32
144) -> IO(Unit):
145  Ez.http.serve_forever_with(~C,
146    ~(c => req => http.answer(C, run, srv, path, origins, c, req)), ctx, host,
147    port, max)
148
149# --- with the server in the context ---
150#
151# A server passed as `~srv` is a closed template too, so one built at run
152# time (from a document read at startup, say) rides in the context: `srv`
153# reads it from the context on every request
154
155# the endpoint's answer to one request ezhttp read, its server read from the
156# context
157def http.answer_from(
158  ~C: Data,
159  ~srv: C -> Server.Server,
160  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
161  ~path: String,
162  ~origins: List<&2, String>,
163  +ctx: C,
164  req: EzServer.Req
165) -> IO(EzServer.Answer):
166  IO.bind(Http.Reply, EzServer.Answer,
167    http.web(C, run, ctx, Web.decide(srv(ctx), path, origins,
168      EzServer.server.plain(req))),
169    EzServer.server.text)
170
171# serve `limit` connections, at most `max` at once (0: no limit), then stop
172def http.serve_from(
173  ~C: Data,
174  ~srv: C -> Server.Server,
175  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
176  ~path: String,
177  ~origins: List<&2, String>,
178  +ctx: C,
179  host: String,
180  port: U32,
181  limit: U32,
182  +max: U32
183) -> IO(Unit):
184  Ez.http.serve_io_with(~C,
185    ~(c => req => http.answer_from(C, srv, run, path, origins, c, req)), ctx,
186    host, port, limit, max)
187
188# serve until the process is stopped, at most `max` connections at once
189# (0: no limit)
190@unsafe def http.serve_forever_from(
191  ~C: Data,
192  ~srv: C -> Server.Server,
193  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
194  ~path: String,
195  ~origins: List<&2, String>,
196  +ctx: C,
197  host: String,
198  port: U32,
199  +max: U32
200) -> IO(Unit):
201  Ez.http.serve_forever_with(~C,
202    ~(c => req => http.answer_from(C, srv, run, path, origins, c, req)), ctx,
203    host, port, max)
204
205# --- with no context ---
206
207# one HTTP request to a server's endpoint at `path`
208def http.handle(
209  ~run: String -> Value.Json -> IO(Rpc.Out),
210  ~srv: Server.Server,
211  ~path: String,
212  ~origins: List<&2, String>,
213  req: Http.Request
214) -> IO(Http.Reply):
215  http.handle_with(~Unit, ~(_c => n => a => run(n, a)), srv, path, origins,
216    Unit{}, req)
217
218# serve `limit` connections, as many at once as arrive, then stop
219def http.serve(
220  ~run: String -> Value.Json -> IO(Rpc.Out),
221  ~srv: Server.Server,
222  ~path: String,
223  ~origins: List<&2, String>,
224  host: String,
225  port: U32,
226  limit: U32
227) -> IO(Unit):
228  http.serve_with(~Unit, ~(_c => n => a => run(n, a)), srv, path, origins,
229    Unit{}, host, port, limit, 0)
230
231# serve until the process is stopped, as many connections at once as arrive
232@unsafe def http.serve_forever(
233  ~run: String -> Value.Json -> IO(Rpc.Out),
234  ~srv: Server.Server,
235  ~path: String,
236  ~origins: List<&2, String>,
237  host: String,
238  port: U32
239) -> IO(Unit):
240  http.serve_forever_with(~Unit, ~(_c => n => a => run(n, a)), srv, path,
241    origins, Unit{}, host, port, 0)
242
243# --- stdio ---
244
245# up to `max` bytes from file descriptor 0, as one read(2) returns them;
246# empty at the end of input (MCP-TRUST-5)
247def ezmcp.stdin(max: U32) -> IO(Result<&1, &1, U32 & String, List<&2, U32>>):
248  import "./effs/stdin.c"
249  import "./effs/stdin.js"
250
251# bytes one read of stdin asks for
252def stdio.chunk() -> U32:
253  65536
254
255# a batch's replies: nothing when there are none, else one line
256def stdio.batch(replies: Maybe<&2, Value.Json>) -> IO(Unit):
257  match replies:
258    case None{}:
259      IO.pure(Unit, Unit{})
260    case Some{batch}:
261      IO.write(Line.frame(batch))
262
263# act on one line's act, read by revision `rev`: write a reply as one line,
264# or run the tool and write its reply, or run a batch's acts and write the
265# array of their replies, or write nothing
266def stdio.act(
267  ~C: Data,
268  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
269  +ctx: C,
270  +rev: String,
271  act: Server.Act
272) -> IO(Unit):
273  match act:
274    case Server.Quiet{}:
275      IO.pure(Unit, Unit{})
276    case Server.Send{reply}:
277      IO.write(Line.frame(reply))
278    case Server.Fault{reply}:
279      IO.write(Line.frame(reply))
280    case Server.Run{id, name, args}:
281      IO.bind(Rpc.Out, Unit, run(ctx, name, args),
282        out => IO.write(Line.frame(Server.ran(rev, id, out))))
283    case Server.Batch{acts}:
284      IO.bind(List<&2, Maybe<&2, Value.Json>>, Unit, act.all(C, run, ctx, rev, acts),
285        rs => stdio.batch(Server.batch.reply(rs)))
286
287# act on a turn read by revision `rev`, then go on with the revision after it
288def stdio.turn(
289  ~C: Data,
290  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
291  +ctx: C,
292  +rev: String,
293  turn: Server.Turn,
294  next: String -> IO(String)
295) -> IO(String):
296  match turn:
297    case Server.Turn{act, after}:
298      IO.bind(Unit, String, stdio.act(C, run, ctx, rev, act), _u => next(after))
299
300# every whole line, in order, each read by the revision the line before it
301# left; answers the revision the last one left
302def stdio.lines(
303  ~C: Data,
304  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
305  ~srv: Server.Server,
306  +ctx: C,
307  lines: List<&2, List<&2, U32>>,
308  +rev: String
309) -> IO(String):
310  match lines:
311    case []:
312      IO.pure(String, rev)
313    case line <> rest:
314      stdio.turn(C, run, ctx, rev, Line.line(srv, rev, line),
315        after => stdio.lines(C, run, srv, ctx, rest, after))
316
317# act on a cut's whole lines, then read on with its rest
318def stdio.cut(
319  ~C: Data,
320  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
321  ~srv: Server.Server,
322  +ctx: C,
323  rev: String,
324  again: List<&2, U32> -> String -> IO(Unit),
325  cut: Line.Cut
326) -> IO(Unit):
327  match cut:
328    case Line.Cut{lines, rest}:
329      IO.bind(String, Unit, stdio.lines(C, run, srv, ctx, lines, rev),
330        after => again(rest, after))
331
332# after a read: an empty one is the end of stdin, and the bytes after the
333# last newline are the last message; any other is cut into lines
334def stdio.chunked(
335  ~C: Data,
336  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
337  ~srv: Server.Server,
338  +ctx: C,
339  rev: String,
340  buf: List<&2, U32>,
341  bytes: List<&2, U32>,
342  again: List<&2, U32> -> String -> IO(Unit),
343  empty: Bool
344) -> IO(Unit):
345  match empty:
346    case True{}:
347      IO.bind(String, Unit, stdio.lines(C, run, srv, ctx, [buf], rev),
348        _r => IO.pure(Unit, Unit{}))
349    case False{}:
350      stdio.cut(C, run, srv, ctx, rev, again,
351        Line.cut(List.append(&2, U32, buf, bytes)))
352
353# after one read: a failure is logged to stderr and ends the loop
354def stdio.got(
355  ~C: Data,
356  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
357  ~srv: Server.Server,
358  +ctx: C,
359  rev: String,
360  buf: List<&2, U32>,
361  again: List<&2, U32> -> String -> IO(Unit),
362  res: Result<&1, &1, U32 & String, List<&2, U32>>
363) -> IO(Unit):
364  match res:
365    case Fail{(_code, why)}:
366      IO.print_err("ezmcp: stdin: " ++ why)
367    case Done{+bytes}:
368      stdio.chunked(C, run, srv, ctx, rev, buf, bytes, again,
369        List.is_empty(&2, U32, bytes))
370
371# read, act on every whole line, and read again until stdin ends; `rev` is
372# the revision the next line is read by
373@unsafe def stdio.loop(
374  ~C: Data,
375  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
376  ~srv: Server.Server,
377  +ctx: C,
378  buf: List<&2, U32>,
379  rev: String
380) -> IO(Unit):
381  IO.bind(Result<&1, &1, U32 & String, List<&2, U32>>, Unit,
382    ezmcp.stdin(stdio.chunk()),
383    res => stdio.got(C, run, srv, ctx, rev, buf,
384      rest => after => stdio.loop(C, run, srv, ctx, rest, after), res))
385
386# serve newline-delimited JSON-RPC on stdin and stdout until stdin ends; a
387# line is read by the latest revision until an initialize names another
388@unsafe def stdio.serve_with(
389  ~C: Data,
390  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
391  ~srv: Server.Server,
392  +ctx: C
393) -> IO(Unit):
394  stdio.loop(C, run, srv, ctx, [], Server.latest())
395
396# serve newline-delimited JSON-RPC on stdin and stdout until stdin ends,
397# with no context
398@unsafe def stdio.serve(
399  ~run: String -> Value.Json -> IO(Rpc.Out),
400  ~srv: Server.Server
401) -> IO(Unit):
402  stdio.serve_with(~Unit, ~(_c => n => a => run(n, a)), srv, Unit{})
403
404# --- client ---
405
406# what an ezhttp response to a request says
407def client.took(res: EzClient.Response) -> Rpc.Reply:
408  match res:
409    case EzClient.Err{why}:
410      Rpc.Unread{why}
411    case EzClient.Response{status, +headers, reason, bytes}:
412      Client.took(status, Http.headers.find(headers, "Content-Type"),
413        Ez.body_text(bytes))
414
415# POST one request and read its reply
416def client.ask(
417  url: String,
418  version: String,
419  timeout_ms: U32,
420  msg: Value.Json
421) -> IO(Rpc.Reply):
422  IO.bind(EzClient.Response, Rpc.Reply,
423    Ez.http.post_opts(url, Client.headers(version), Ez.text_body(Json.print(msg)),
424      timeout_ms),
425    r => IO.pure(Rpc.Reply, client.took(r)))
426
427def client.sent(res: EzClient.Response) -> Bool:
428  match res:
429    case EzClient.Err{why}:
430      False{}
431    case EzClient.Response{status, headers, reason, bytes}:
432      Client.accepted(status)
433
434# POST one notification; whether the server accepted it
435def client.tell(
436  url: String,
437  version: String,
438  timeout_ms: U32,
439  msg: Value.Json
440) -> IO(Bool):
441  IO.bind(EzClient.Response, Bool,
442    Ez.http.post_opts(url, Client.headers(version), Ez.text_body(Json.print(msg)),
443      timeout_ms),
444    r => IO.pure(Bool, client.sent(r)))