AI Agents in Go

Run itExample

Streaming an agent’s progress to a browser

A server written with net/http alone sends an agent’s events, as they happen, to a page that reads them with EventSource. The headers, the flush, a deadline per write, a heartbeat, a client that leaves, and a shutdown that ends open streams, from a tested example.

Draws on chapter 6 of the bookUpdated

Chapter 6 of the book reads server-sent events from the model’s API and turns them into the agent’s own events: text as it arrives, each tool call, each result, the end of a turn, and the end of the run. This example sends those events onward, from a Go server to a web page, so a person can watch a run as it happens and stop it. It is the companion code’s ssebrowser example, built on net/http alone, with tests that run under the race detector.

The model is scripted, a replay of chapter 1’s run on the duration project slowed down so you can watch it, so everything runs offline. The tools, go test among them, run for real.

Run it

From the code directory of a clone of the companion repository:

$ go run ./ssebrowser

Open the address it prints and press Run. The page shows the model’s text as it streams, each tool call and its result, and the end of each turn. Press Stop partway through and the run on the server stops too. To see the raw stream without a browser:

$ curl -N 'http://localhost:8080/events?task=fix'

The page sends its task in the query string, but the scripted model gives the same replies whatever the task says.

One request, one stream

The event stream is a long-lived GET, since a browser’s EventSource can’t send anything else. Here is the handler:

ssebrowser/server.go
// events streams one run's events to the client. The task comes in the
// query string, since a browser's EventSource can only send a GET.
func (s *Server) events(w http.ResponseWriter, r *http.Request) {
	task := r.URL.Query().Get("task")
	if task == "" {
		http.Error(w, "missing task", http.StatusBadRequest)
		return
	}
	h := w.Header()
	h.Set("Content-Type", "text/event-stream")
	h.Set("Cache-Control", "no-cache")
	h.Set("X-Accel-Buffering", "no")
	st := &stream{w: w, rc: http.NewResponseController(w),
		wait: s.WriteWait}
	ctx, cancel := context.WithCancel(r.Context())
	events := s.start(ctx, task)
	defer func() {
		cancel()
		for range events {
		}
	}()
	if err := s.pump(st, events); err != nil {
		slog.DebugContext(ctx, "event stream ended", "err", err)
	}
}

Three headers make it a stream: text/event-stream tells the browser what it is reading, no-cache keeps caches out of the way, and X-Accel-Buffering: no asks a proxy such as nginx not to hold events back in a buffer.

The run gets a context derived from the request’s. When the browser closes the connection, because the person pressed Stop or closed the tab, the request’s context ends, and the run’s model call and tools stop with it. The deferred function cancels the run and then reads the channel until it closes, so the goroutine that drives the agent has always finished before the handler returns. That goroutine has one owner, the handler, and one way to stop, the context.

The agent runs in its own goroutine for a reason: while the model is thinking, nothing arrives for a while, and the handler has to keep sending something. pump, in the same file, waits on two things at once, the next event and a ticker, and sends a comment line, : heartbeat, when the ticker fires first. Proxies and load balancers may close a connection that stays silent too long; the heartbeat keeps it open, and EventSource ignores comment lines. Only the handler writes to the client, because a ResponseWriter isn’t safe to use from two goroutines.

Every write is flushed, with a deadline

ssebrowser/server.go
// stream writes an event stream to one client.
type stream struct {
	w    io.Writer
	rc   *http.ResponseController
	wait time.Duration
}

// send writes msg and flushes it to the client at once. A client that
// doesn't take it within s.wait is dropped: the write fails, and the
// handler stops.
func (s *stream) send(msg string) error {
	err := s.rc.SetWriteDeadline(time.Now().Add(s.wait))
	if err != nil {
		return fmt.Errorf("set write deadline: %w", err)
	}
	if _, err := io.WriteString(s.w, msg); err != nil {
		return fmt.Errorf("write event: %w", err)
	}
	if err := s.rc.Flush(); err != nil {
		return fmt.Errorf("flush event: %w", err)
	}
	return nil
}

http.ResponseController gives a handler two things a plain ResponseWriter hides. Flush pushes each event to the client at once instead of letting it wait in a buffer, which is the whole point of a stream. SetWriteDeadline sets a limit on each write: a client that stops reading, with the connection still open, would otherwise hold its goroutine and its run forever. Past the deadline the write fails, pump returns, and the deferred cancel stops the run. The program allows ten seconds per write and sends a heartbeat every fifteen.

The wire format

Each event is a line naming its kind, a line of JSON, and a blank line:

ssebrowser/sse.go
// encode formats ev as one event in the stream: a line naming its kind,
// a line of JSON data, and a blank line that ends the event. JSON
// escapes newlines, so the data always fits on its one line.
func encode(ev agent.Event) (string, error) {
	name, d, err := wire(ev)
	if err != nil {
		return "", err
	}
	b, err := json.Marshal(d, jsontext.AllowInvalidUTF8(true))
	if err != nil {
		return "", fmt.Errorf("encode %s event: %w", name, err)
	}
	return fmt.Sprintf("event: %s\ndata: %s\n\n", name, b), nil
}

Newlines inside the data would end the event early, and JSON escapes them, so a tool result of any length fits on its single data: line. Naming each event lets the page attach one listener per kind instead of parsing a type field out of every message.

The page

The page reads the stream with EventSource, the browser’s built-in client for server-sent events:

ssebrowser/index.html
run.onclick = () => {
  log.replaceChildren();
  text = null;
  const task = document.getElementById("task").value;
  es = new EventSource("/events?task=" + encodeURIComponent(task));
  run.disabled = true;
  stop.disabled = false;
  const on = (kind, f) =>
    es.addEventListener(kind, e => f(JSON.parse(e.data)));
  on("text", d => {
    text ??= line("text", "");
    text.textContent += d.text;
  });
  on("call", d => {
    text = null;
    line("call", "→ " + d.name + " " + d.args);
  });
  on("result", d =>
    line(d.is_error ? "result error" : "result", d.content));
  on("turn", d => {
    text = null;
    line("turn meta", "end of turn " + d.turn);
  });
  on("stopped", d => d.error
    ? done("stopped: " + d.error, "meta error")
    : done("answered", "meta"));
  es.onerror = () => done("connection lost", "meta error");
};

One detail matters more than it looks. When the stream ends, EventSource reconnects by default, and here a reconnection would start a whole new run. So the page closes the EventSource itself when it sees the stopped event or an error, and the Stop button does the same.

Shutting down

A stream never goes idle by itself, so http.Server.Shutdown, which waits for active connections to go idle, would wait out its whole timeout. The program gives the server a BaseContext that returns the program’s own context, which Ctrl-C cancels:

ssebrowser/main.go
// serve runs the server until ctx is done, then shuts it down.
func serve(ctx context.Context, addr string, w io.Writer) error {
	s := &Server{
		Run:       demo(700*time.Millisecond, 40*time.Millisecond),
		Heartbeat: 15 * time.Second,
		WriteWait: 10 * time.Second,
	}
	srv := &http.Server{
		Handler:           s.Handler(),
		ReadHeaderTimeout: 5 * time.Second,
		BaseContext: func(net.Listener) context.Context {
			return ctx
		},
	}
	ln, err := new(net.ListenConfig).Listen(ctx, "tcp", addr)
	if err != nil {
		return fmt.Errorf("listen: %w", err)
	}
	_, err = fmt.Fprintf(w, "open http://%s\n", ln.Addr())
	if err != nil {
		return fmt.Errorf("print address: %w", err)
	}

Every request’s context descends from it, so cancelling it ends every open stream and its run, and Shutdown then finds nothing left to wait for. ReadHeaderTimeout bounds how long a client may take to send its headers, a limit worth setting on any server that faces the internet.

What the tests check

$ go test -race ./ssebrowser

The tests read the stream from an httptest server and check the headers and every event of a full run, that a quiet stream sends heartbeats, that a client leaving cancels the run, that a failed write stops the handler, that bad requests are refused, and that shutdown ends open streams.

What a real service adds

The example is small on purpose, and its README lists what a public service would change:

  • Error text. The page shows a run’s error as it is; a public service would send a short stop reason instead of internal detail.
  • Limits. Each request starts a run. A real service caps how many run at once, and per user, since each one spends money.
  • Reconnecting. A dropped connection loses the run. To let a client pick up where it left off, give each event an id: line, keep the run going apart from the request, and resume from the Last-Event-ID header the browser sends when it reconnects.

Key files