ajet/tbd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

framework-ify. proper agent runtime + adopt ag-ui
CI / Test + coverage (pull_request) Successful in 1m41s
Details

Browse Source
This commit is contained in:
Adam Jeniski
2026-09-21 18:34:19 -09:00
parent 187252177c
commit a3ef6ec077
26 changed files with 1514 additions and 1200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/tbd/agent/a2a.clj
+182
View File
@@ -0,0 +1,182 @@
(ns tbd.agent.a2a
"The A2A (Agent2Agent, https://a2a-protocol.org) server, mounted at `/a2a`, with its discovery
document at `/.well-known/agent-card.json`. Unlike /api/chat (the in-page chatbot, driven by the
browser's own tools) an A2A caller has no page open, so this agent's tools are the app's
server-side ones — the same operations tbd.mcp.tools exposes to MCP clients, defined again here
directly over tbd.todo.service/tbd.math.core (not through plumcp's tool-item wrapper, which
tbd.mcp.tools builds for a different transport) so this ns has no plumcp dependency.
A2A calls are synchronous (`message/send` only: no `message/stream`, and the AgentCard says so
truthfully in `capabilities.streaming`) — /api/chat and the AG-UI protocol are where this app's
streaming surface is. Since there is no browser to run a tool and report back, this ns runs the
tool loop itself: ask the agent, and if it asks for tools, run them immediately and ask again,
until it answers in text or `max-turns` is reached.
Protected the same way /mcp is (tbd.oauth.middleware/wrap-bearer-auth): any access token this
server issued works here too, since verifying one doesn't check which `resource` it was minted
for (see tbd.oauth.middleware) — a client that has authorized for MCP can use A2A with the same
token."
(:require [clojure.string :as str]
[malli.json-schema :as json-schema]
[muuntaja.core :as m]
[tbd.agent.core :as agent]
[tbd.agent.instructions :as instructions]
[tbd.math.core :as math]
[tbd.math.schema :as math-schema]
[tbd.mcp.schema :refer [coerce!]]
[tbd.oauth.middleware :as oauth-mw]
[tbd.todo.schema :as todo-schema]
[tbd.todo.service :as todo]))
(def ^:private max-turns
"How many rounds of tool calls the agent may make before answering, for one A2A message."
8)
;; --- Tools: the same operations tbd.mcp.tools offers MCP clients, for a caller with no page ---
(defn- ok! [{:keys [ok error]}]
(if error (throw (ex-info error {})) ok))
(defn- ->json [data] (slurp (m/encode "application/json" data)))
(defn- tool [name description schema handler]
{:name name :description description
:inputSchema (json-schema/transform schema)
:handler (fn [args {:keys [user ds]}] (str (handler (coerce! schema args) user ds)))})
(def tools
[(tool "add" "Add two numbers" math-schema/Add
(fn [{:keys [x y]} _ _] (math/add x y)))
(tool "whoami" "Says which user this call acts as, and whether they are an admin" [:map]
(fn [_ {:keys [login id admin?]} _]
(format "%s (Gitea user %d)%s" login id (if admin? ", admin" ""))))
(tool "list_todos" "Lists your todos (JSON: id, title, done, created-at, updated-at), oldest first" [:map]
(fn [_ user ds] (->json (ok! (todo/list-todos ds user)))))
(tool "get_todo" "Gets one of your todos by id (JSON)" todo-schema/IdParam
(fn [{:keys [id]} user ds] (->json (ok! (todo/find-todo ds user id)))))
(tool "create_todo" "Creates a todo for you and returns it (JSON)" todo-schema/NewTodo
(fn [{:keys [title]} user ds] (->json (ok! (todo/create! ds user title)))))
(tool "update_todo" "Changes the title and/or done flag of one of your todos and returns it (JSON)"
todo-schema/UpdateTodo
(fn [{:keys [id] :as input} user ds] (->json (ok! (todo/update! ds user id (dissoc input :id))))))
(tool "delete_todo" "Deletes one of your todos" todo-schema/IdParam
(fn [{:keys [id]} user ds] (ok! (todo/delete! ds user id)) "deleted"))])
(defn- execute-tool [ctx {:keys [name arguments]}]
(if-let [{:keys [handler]} (first (filter #(= name (:name %)) tools))]
(try (handler (m/decode "application/json" (or arguments "{}")) ctx)
(catch Exception e (str "error: " (ex-message e))))
(str "error: no such tool " (pr-str name))))
;; --- Driving the agent, running its own tool loop (no browser here) ---
(defn- collect
"Drains `pump` (see tbd.agent.core/start), returning {:text s} or {:tool-calls [...]} or
{:error s}: exactly one of them, whichever the turn ended with."
[pump]
(let [text (StringBuilder.) result (atom nil)]
(pump (fn [{:keys [content tool-calls done error]}]
(cond
content (.append text ^String content)
tool-calls (reset! result {:tool-calls tool-calls})
error (reset! result {:error error})
done (when-not @result (reset! result {:text (.toString text)})))))
@result))
(defn- run-tools!
"Answers `messages` (tbd.agent.core's internal Message shape), running any tool calls the agent
makes itself, `ctx` ({:user :ds}) being what the tools need. {:text s} or {:error s :kind ...}."
[config ctx messages]
(loop [messages messages turn 0]
(if (>= turn max-turns)
{:error "the agent didn't reach an answer in time" :kind :upstream}
(let [{:keys [ok error kind]} (agent/start config (instructions/a2a) messages tools)]
(if error
{:error error :kind kind}
(let [{:keys [text tool-calls error]} (collect ok)]
(cond
error {:error error :kind :upstream}
(seq tool-calls)
(recur (into (conj (vec messages) {:role "assistant" :content "" :tool-calls tool-calls})
(map (fn [{:keys [id name arguments]}]
{:role "tool" :tool-call-id id
:content (execute-tool ctx {:name name :arguments arguments})}))
tool-calls)
(inc turn))
:else {:text text})))))))
(defn- message-text
"The plain text of an A2A Message's `parts` ([{kind: \"text\" text: \"...\"}, ...]), joined."
[parts]
(->> parts (filter #(= "text" (or (:kind %) (get % "kind")))) (map #(or (:text %) (get % "text"))) (str/join "\n")))
(defn- agent-message
"A reply as an A2A Message: role \"agent\", one TextPart."
[text]
{:role "agent" :parts [{:kind "text" :text text}] :messageId (str (random-uuid)) :kind "message"})
;; --- JSON-RPC over HTTP ---
(defn- json [status body]
{:status status
:headers {"Content-Type" "application/json" "Cache-Control" "no-store"}
:body (m/encode "application/json" body)})
(defn- rpc-result [id result] {:jsonrpc "2.0" :id id :result result})
(defn- rpc-error [id code message] {:jsonrpc "2.0" :id id :error {:code code :message message}})
(def ^:private error-status {:unavailable 503 :upstream 502})
(defn- handle-message-send [config ctx {:keys [message]}]
(let [text (message-text (:parts message))]
(if (str/blank? text)
{:jsonrpc-error [-32602 "message has no text parts"]}
(let [{:keys [text error kind]} (run-tools! config ctx [{:role "user" :content text}])]
(if error
{:error [(get error-status kind 502) error]}
{:result (agent-message text)})))))
(defn- handler [config ds]
(fn [{:keys [user body]}]
(let [{:keys [jsonrpc id method params]} (m/decode "application/json" (slurp body))]
(cond
(not= "2.0" jsonrpc)
(json 200 (rpc-error id -32600 "invalid JSON-RPC request"))
(= "message/send" method)
(let [{:keys [result jsonrpc-error error]} (handle-message-send config {:user user :ds ds} params)]
(cond
result (json 200 (rpc-result id result))
jsonrpc-error (json 200 (apply rpc-error id jsonrpc-error))
:else (let [[status text] error] (json status (rpc-error id -32000 text)))))
:else (json 200 (rpc-error id -32601 (str "method not found: " method)))))))
(defn- agent-card [base-url]
{:name "tbd"
:description "tbd's agent: todo management and simple math, on behalf of the user who authorized this connection."
:url (str base-url "/a2a")
:version "1.0.0"
:protocolVersion "0.3.0"
:preferredTransport "JSONRPC"
:capabilities {:streaming false :pushNotifications false}
:defaultInputModes ["text/plain"]
:defaultOutputModes ["text/plain"]
:securitySchemes {"bearer" {:type "http" :scheme "bearer"
:description "An OAuth access token from this server's own authorization server (see /.well-known/oauth-protected-resource/mcp; the same token works here)."}}
:security [{"bearer" []}]
:skills (mapv (fn [{:keys [name description]}]
{:id name :name name :description description :tags ["tbd"]})
tools)})
(defn routes
"`auth` is the auth config (for `:base-url` and the OAuth check), `ds` a delay of the app's
database, `chat` the agent config (see tbd.agent.core/env-config)."
[auth ds chat]
["" {:no-doc true}
["/.well-known/agent-card.json"
{:middleware ^:replace []
:get {:handler (fn [_req] (json 200 (agent-card (:base-url auth))))}}]
["/a2a"
{:middleware ^:replace []
:post {:handler (-> (handler chat ds) (oauth-mw/wrap-bearer-auth auth ds))}}]])
src/tbd/agent/core.clj
+299
View File
@@ -0,0 +1,299 @@
(ns tbd.agent.core
"The chatbot's agent: a LangChain4j `StreamingChatModel` talking to the OpenAI-compatible LM
Studio server named by the environment. No Ring here; tbd.chat.routes owns that, tbd.agent.a2a
drives the same model for the A2A server.
`start` sends the request and gives {:ok pump} or {:error text :kind :unavailable|:upstream},
like tbd.todo.service's. `pump` forwards the reply, as it is generated, to a function that takes
events: {:thinking \"text\"} (a piece of the model's reasoning), {:content \"text\"} (a piece of
the reply), {:tool-calls [{:id :name :arguments}]} (the model wants these tools run, once its
reply is complete: `arguments` is the JSON text the model wrote), and finally {:done true} or
{:error \"text\"}. This event shape is internal (tbd.chat.routes and tbd.agent.a2a each translate
it to their own wire protocol — AG-UI events, A2A artifacts) and is what tbd.chat.trace watches.
The system prompt is the agent's own (tbd.agent.instructions), not something the caller supplies:
`start` takes `system-text` already composed (tbd.agent.instructions/chat-system-message for the
in-page chatbot, tbd.agent.instructions/a2a for the A2A server), and always sends it as the first
message. `tools` are MCP-style {:name :description :inputSchema}, converted to LangChain4j's
typed tool specifications (tbd.agent.json-schema) — the model only sees their metadata; running
them and reporting the result back (as a \"tool\" role message, see tbd.chat.schema/Message) is
the caller's job.
Every turn is traced (see tbd.chat.trace): what was asked, how the reply went, and whether the
model looks stuck in a loop. The config's `:trace` says where that goes."
(:require [clojure.string :as str]
[tbd.agent.json-schema :as json-schema]
[tbd.chat.trace :as trace])
(:import (dev.langchain4j.agent.tool ToolExecutionRequest ToolSpecification)
(dev.langchain4j.data.message AiMessage ChatMessage SystemMessage ToolExecutionResultMessage UserMessage)
(dev.langchain4j.http.client.jdk JdkHttpClientBuilder)
(dev.langchain4j.model.chat StreamingChatModel)
(dev.langchain4j.model.chat.request ChatRequest)
(dev.langchain4j.model.chat.response
ChatResponse CompleteToolCall PartialResponse PartialResponseContext
PartialThinking PartialThinkingContext PartialToolCallContext
StreamingChatResponseHandler StreamingHandle)
(dev.langchain4j.model.openai OpenAiStreamingChatModel)
(java.net.http HttpClient HttpClient$Version)
(java.time Duration)
(java.util.concurrent LinkedBlockingQueue ScheduledFuture ScheduledThreadPoolExecutor ThreadFactory TimeUnit)))
(def default-timeout-seconds 120)
(defn- env-val [env k]
(let [v (get env k)]
(when-not (str/blank? v) (str/trim v))))
;; HTTP/1.1 on purpose, same reason as the old tbd.chat.core: the JDK client's default is to ask a
;; plain-http server to upgrade to HTTP/2 (`Upgrade: h2c`), and LM Studio's server never answers
;; that request, so the backend would wait out the whole timeout while the LLM sat idle.
(defn- jdk-http-client-builder []
(-> (JdkHttpClientBuilder.)
(.httpClientBuilder (-> (HttpClient/newBuilder) (.version HttpClient$Version/HTTP_1_1)))))
(defn- api-base-url
"`url` (the server's address, no path, see env-config) with the `/v1` LM Studio's OpenAI-compatible
endpoints live under: LangChain4j's OpenAI client posts straight to `<baseUrl>/chat/completions`,
it does not add this itself (its own default base url is `https://api.openai.com/v1`). Without
it, LM Studio answers 200 with a JSON error body langchain4j can't tell from an empty reply, so
every turn silently becomes \"the LLM server sent no reply\" instead of a clear error."
[url]
(if (str/ends-with? url "/v1") url (str url "/v1")))
(defn- build-chat-model ^StreamingChatModel [{:keys [url model api-key timeout-seconds]}]
(let [builder (-> (OpenAiStreamingChatModel/builder)
(.httpClientBuilder (jdk-http-client-builder))
(.baseUrl (api-base-url url))
(.apiKey (or api-key "not-needed"))
(.timeout (Duration/ofSeconds timeout-seconds))
;; LM Studio's reasoning models send `reasoning_content`; this is what maps it
;; onto onPartialThinking instead of dropping it.
(.returnThinking true))]
(when model (.modelName builder model))
(.build builder)))
(defn env-config
"Chat config from environment variables:
LMSTUDIO_URL where the LM Studio server is, e.g. http://192.168.1.20:1234 (chat is off without it)
LMSTUDIO_MODEL the model id to ask for (optional: LM Studio uses the model that is loaded)
LMSTUDIO_API_KEY sent as a bearer token, when the server wants one (optional)
LMSTUDIO_TIMEOUT_SECONDS how long the server may stay silent before giving up (default 120)
CHAT_TRACE_FILE a file to append a JSON trace of every turn to, with the model's full reasoning and
replies and the messages it was sent (optional: without it, each turn is only
summarised in the log)"
([] (env-config (System/getenv)))
([env]
(let [config {:url (some-> (env-val env "LMSTUDIO_URL") (str/replace #"/+$" ""))
:model (env-val env "LMSTUDIO_MODEL")
:api-key (env-val env "LMSTUDIO_API_KEY")
:timeout-seconds (or (some-> (env-val env "LMSTUDIO_TIMEOUT_SECONDS") parse-long)
default-timeout-seconds)
:trace (trace/config-for (env-val env "CHAT_TRACE_FILE"))}]
(assoc config :chat-model-delay (delay (build-chat-model config))))))
(defn configured? [{:keys [url]}] (some? url))
(defn- chat-model
"The config's `StreamingChatModel`: `:chat-model` directly, when a caller (a test) gives one, else
the one built from `:url`/`:model`/`:api-key`/`:timeout-seconds`, built once and reused."
^StreamingChatModel [config]
(or (:chat-model config) (force (:chat-model-delay config))))
;; --- Building the request ---
(defn- llm-message
"One message of the conversation (see tbd.chat.schema/Message) as a LangChain4j ChatMessage."
^ChatMessage [{:keys [role content tool-calls tool-call-id]}]
(case role
"user" (UserMessage/from ^String content)
"assistant" (if (seq tool-calls)
(AiMessage/from (str content)
^java.util.List
(mapv (fn [{:keys [id name arguments]}]
(-> (ToolExecutionRequest/builder)
(.id id) (.name name) (.arguments arguments)
.build))
tool-calls))
(AiMessage/from ^String content))
"tool" (ToolExecutionResultMessage/toolExecutionResultMessage
^String tool-call-id "" ^String content)))
(defn- llm-tool
"An MCP tool's metadata as a function the LLM may call."
^ToolSpecification [{:keys [name description inputSchema]}]
(let [builder (-> (ToolSpecification/builder)
(.name name)
(.parameters (json-schema/tool-parameters inputSchema)))]
(when description (.description builder description))
(.build builder)))
(defn- chat-request ^ChatRequest [system-text messages tools]
(let [builder (-> (ChatRequest/builder)
(.messages ^java.util.List (into [(SystemMessage/from ^String system-text)]
(map llm-message) messages)))]
(when (seq tools)
(.toolSpecifications builder ^java.util.List (mapv llm-tool tools)))
(.build builder)))
;; --- Reading the reply ---
(def ^:private watchdog
(delay (doto (ScheduledThreadPoolExecutor.
1 (reify ThreadFactory
(newThread [_ r] (doto (Thread. ^Runnable r "chat-watchdog") (.setDaemon true)))))
(.setRemoveOnCancelPolicy true))))
(defn- exception-detail [^Throwable e]
(str (class e) ": " (.getMessage e)))
(defn- fail
([text] (fail text nil))
([text detail]
(println (cond-> (str "[tbd] chat: " text) detail (str " (" detail ")")))
text))
(defn- streaming-handler
"A StreamingChatResponseHandler that pushes every callback onto `queue` as a map: {:content s},
{:thinking s}, {:tool-call {:id :name :arguments}} (once a call is fully assembled — LangChain4j
does the delta accumulation itself), {:complete response} or {:error throwable}. `handle!` is
called with the first StreamingHandle seen, so the caller can cancel a stuck stream."
^StreamingChatResponseHandler [^LinkedBlockingQueue queue handle!]
(let [capture! (fn [^StreamingHandle h] (when h (handle! h)))]
;; No type hints on the reify methods' params below: hinting an overloaded interface method's
;; params (onPartialResponse/onPartialThinking/onPartialToolCall each have a 1- and a 2-arg
;; overload) confuses Clojure's overload resolution (`NewInstanceMethod.parse`), which throws
;; "Mismatched return type" instead of picking the right one. Arity alone disambiguates fine.
(reify StreamingChatResponseHandler
(onPartialResponse [_ partial ctx]
(capture! (.streamingHandle ^PartialResponseContext ctx))
(.put queue {:content (.text ^PartialResponse partial)})
nil)
(onPartialThinking [_ partial ctx]
(capture! (.streamingHandle ^PartialThinkingContext ctx))
(.put queue {:thinking (.text ^PartialThinking partial)})
nil)
(onPartialToolCall [_ _partial ctx]
(capture! (.streamingHandle ^PartialToolCallContext ctx))
nil)
(onCompleteToolCall [_ complete]
(let [req (.toolExecutionRequest ^CompleteToolCall complete)]
(.put queue {:tool-call {:id (.id req) :name (.name req) :arguments (.arguments req)}}))
nil)
(onCompleteResponse [_ response]
(.put queue {:complete response})
nil)
(onError [_ error]
(.put queue {:error error})
nil))))
(defn- usage-of [^ChatResponse response]
(when-let [usage (.tokenUsage response)]
{:prompt_tokens (.inputTokenCount usage)
:completion_tokens (.outputTokenCount usage)
:total_tokens (.totalTokenCount usage)}))
(defn- finish-reason-of [^ChatResponse response]
(some-> (.finishReason response) str str/lower-case))
(defn- pump
"Drains `queue`, starting from the already-polled `first-item` (see `start`, which peeks one item
before deciding {:ok}/{:error}), calling `emit` with each event (see the ns docstring).
`timeout-seconds` of silence between items is an error, closing the stream via `handle-atom` (a
StreamingHandle, once one has arrived). An exception from `emit` (the client went away, say) ends
it too, closing the stream, and is rethrown. `tracer` (see tbd.chat.trace) sees everything the
LLM says, and how the turn ended, before the client is told."
[^LinkedBlockingQueue queue handle-atom timeout-seconds emit tracer first-item]
(let [timer (atom nil)
timed-out? (atom false)
cancel-stream! (fn []
(when-let [^StreamingHandle h @handle-atom]
(try (.cancel h) (catch Exception _ nil))))
arm! (fn []
(some-> ^ScheduledFuture @timer (.cancel false))
(reset! timer (.schedule ^ScheduledThreadPoolExecutor @watchdog
^Runnable (fn [] (reset! timed-out? true) (cancel-stream!))
(long timeout-seconds) TimeUnit/SECONDS)))]
(arm!)
(try
(loop [item first-item
replied? false
calls []]
(cond
(:error item)
(let [^Throwable e (:error item)
text (if @timed-out?
(fail "the LLM server took too long to answer" (str timeout-seconds "s of silence"))
(fail "the LLM server can't be reached or refused" (exception-detail e)))]
(trace/finish! tracer {:outcome :error :error text :detail (exception-detail e)})
(emit {:error text}))
(:complete item)
(let [^ChatResponse response (:complete item)
ai (.aiMessage response)
calls (if (seq calls) calls
(mapv (fn [^ToolExecutionRequest r]
{:id (.id r) :name (.name r) :arguments (.arguments r)})
(.toolExecutionRequests ai)))
;; A provider that never streamed tokens (batched the whole reply) has its text
;; only here, never through :content items above.
batched-text (when (and (not replied?) (empty? calls) (not (str/blank? (.text ai))))
(.text ai))
event {:finish-reason (finish-reason-of response) :usage (usage-of response)}]
(trace/observe! tracer event)
(cond
(seq calls) (do (trace/finish! tracer {:outcome :tool-calls :calls calls})
(emit {:tool-calls calls})
(emit {:done true}))
batched-text (do (trace/finish! tracer {:outcome :done})
(emit {:content batched-text})
(emit {:done true}))
(or replied? (not (str/blank? (.text ai))))
(do (trace/finish! tracer {:outcome :done}) (emit {:done true}))
:else (let [text (fail "the LLM server sent no reply")]
(trace/finish! tracer {:outcome :error :error text})
(emit {:error text}))))
:else
(do (arm!)
(trace/observe! tracer (select-keys item [:content :thinking]))
(when-let [out (not-empty (select-keys item [:content :thinking]))] (emit out))
(recur (.take queue)
(or replied? (contains? item :content))
(cond-> calls (:tool-call item) (conj (:tool-call item)))))))
(finally
(trace/finish! tracer {:outcome :aborted})
(some-> ^ScheduledFuture @timer (.cancel false))
(cancel-stream!)))))
(defn start
"Sends the LLM the conversation `messages` (see tbd.chat.schema/Message), and the `tools` it may
ask the caller to run (MCP's {:name :description :inputSchema}; none if nil), after `system-text`
(the agent's own system prompt, see tbd.agent.instructions — composed by the caller, not the
browser). Waits for the first sign of life from the model (a piece of its reply, or an error) so
that an LLM that is down or refuses is still a clean {:error text :kind :unavailable|:upstream},
not an error inside a stream that has already begun — the same contract the old hand-rolled HTTP
client gave, adapted to LangChain4j's async callbacks (which have no separate \"got the HTTP
response\" moment to block on the way a blocking HTTP client's `.send()` did). Otherwise {:ok
pump}: `(pump emit)` blocks until the reply is complete, calling `emit` with its events (see the
ns docstring). If `emit` throws, the request to the LLM is closed (which stops it generating) and
the exception is rethrown."
[{:keys [timeout-seconds trace] :as config} system-text messages tools]
(if-not (configured? config)
{:error "chat is not configured" :kind :unavailable}
(let [tracer (trace/begin! trace {:messages messages :tools tools :model (:model config)})
upstream-error (fn [text detail]
(fail text detail)
(trace/finish! tracer {:outcome :upstream-error :error text :detail detail})
{:error text :kind :upstream})]
(try
(let [queue (LinkedBlockingQueue.)
handle-atom (atom nil)
handler (streaming-handler queue #(reset! handle-atom %))]
(.chat (chat-model config) (chat-request system-text messages tools) handler)
(if-let [first-item (.poll queue (long timeout-seconds) TimeUnit/SECONDS)]
(if (:error first-item)
(upstream-error "the LLM server can't be reached or refused" (exception-detail (:error first-item)))
{:ok (fn [emit] (pump queue handle-atom timeout-seconds emit tracer first-item))})
(upstream-error "the LLM server took too long to answer" (str timeout-seconds "s of silence"))))
(catch Exception e
(upstream-error "the LLM server can't be reached" (exception-detail e)))))))
src/tbd/agent/instructions.clj
+64
View File
@@ -0,0 +1,64 @@
(ns tbd.agent.instructions
"The system prompts of the two agents tbd.agent.core drives: `chat`, for the in-page chatbot
behind /api/chat, and `a2a`, for the A2A server (tbd.agent.a2a) other agents call. Server-only
on purpose: unlike tbd.agent.prompt (a .cljc the browser bundle also needs, for WebMCP's
`guide`/`task`/`tic-tac-toe`), this ns is never on ui/shadow-cljs.edn's `:source-paths`, so the
chatbot's own instructions can never end up in the compiled UI. It reuses the wording that is
genuinely shared (`act`, `work-it-out`, `tic-tac-toe-rules`, `examples`) from tbd.agent.prompt
rather than duplicating it."
(:require [clojure.string :as str]
[tbd.agent.prompt :as prompt]))
(defn chat
"The in-page chatbot's system prompt, the start of its system message: tbd.agent.core adds
where the user is and the tools it has. Its calls show under its message, so it answers in a
sentence and never repeats them."
[]
(str "You are the assistant built into this web app. Your tools type, click and read on the page the\n"
"user is looking at, and every call shows on their screen the moment you make it.\n"
"\n"
prompt/act
(prompt/work-it-out (str "- Chain calls without stopping. The tool you need belongs to another page: navigate there, then\n"
" use it. You need an id: list first, then act. Never invent an id, a title or the state of the\n"
" page: read it with a tool.\n"))
"Then say one short plain sentence: \"Added it.\" \"That's 23.\" The call and its result are already\n"
"on screen under your message, so never repeat them, never paste JSON, never use markdown,\n"
"headings or bullet lists. Say something was done only if a tool result says it was.\n"
"\n"
"Some things are not tool work: greetings, questions about you, questions you can simply answer.\n"
"Answer those in a sentence or two, with no tool call. Asked what you can do, say it plainly from\n"
"the tools you have. Think briefly before you act.\n"
"\n"
prompt/tic-tac-toe-rules
prompt/examples))
(defn chat-system-message
"`chat`'s prompt, with, last, where the user is (`page`) and the names of the `tools` on offer
there. The tools' names are in the request's `tools` too, but a model picks them more readily
when the instructions say them as well, and it can say what it can do without a call. One
message, and the first: several chat templates (Gemma's, Mistral's) refuse more than one system
message, or one that isn't first."
[page tools]
(cond-> (chat)
page (str "\n\nThe user is on the \"" page "\" page of the app: what it shows is what they can see.")
(seq tools) (str (if page " Its" "\n\nYour")
" tools: " (str/join ", " (map :name tools))
". The tools of other pages are offered once you navigate there.")))
(defn a2a
"The A2A server's system prompt: an outside agent calling this app directly, not through a
page, so it has no DOM to act on and no visual feedback to lean on. Its tools are the app's
server-side ones (tbd.mcp.tools, the same the MCP server offers), not a page's."
[]
(str "You are tbd's agent, reachable over the Agent2Agent protocol. You act on behalf of the user\n"
"who authorized the caller, using the tools you were given: todo management and anything else\n"
"this server exposes as a tool. You have no page open and no visual feedback beyond what a tool\n"
"returns.\n"
"\n"
prompt/act
(prompt/work-it-out (str "- You need an id: list first, then act. Never invent an id, a title or the state of the data:\n"
" read it with a tool.\n"))
"Then report in one short plain sentence, from the tool results: \"Added it.\" \"That's 23.\" Don't\n"
"paste JSON, don't describe the tool call itself, just its outcome.\n"
"\n"
prompt/examples))
src/tbd/agent/json_schema.clj
+80
View File
@@ -0,0 +1,80 @@
(ns tbd.agent.json-schema
"Translates a tool's `inputSchema` (plain JSON Schema, decoded to a Clojure map — see
tbd.chat.schema/Tool) into LangChain4j's typed `JsonSchemaElement` tree, which is what
`ToolSpecification.Builder/parameters` needs (dev.langchain4j.model.chat.request.json). The
browser is the one authoring these schemas (tbd.ui.webmcp), so this only has to cover the shapes
MCP-style tool schemas actually use: object/string/number/integer/boolean/array/enum, nested
arbitrarily. Anything else (a bare `:enum` without a `:type`, an unrecognized `:type`) falls back
to a permissive `JsonRawSchema` dump of the schema as JSON, rather than failing the request."
(:require [clojure.string :as str]
[muuntaja.core :as m])
(:import (dev.langchain4j.model.chat.request.json
JsonArraySchema JsonBooleanSchema JsonEnumSchema JsonIntegerSchema
JsonNumberSchema JsonObjectSchema JsonRawSchema JsonSchemaElement JsonStringSchema)))
(defn- field
"`m`'s value at `k`, trying the keyword first (how muuntaja decodes JSON) then the string, so
this works whether or not something upstream keywordized the nested schema too."
[m k]
(or (get m (keyword k)) (get m (name k))))
(defn- raw
"A schema element that isn't one of the shapes below: dumped as JSON and wrapped in
`JsonRawSchema` rather than dropped, so the LLM still sees it."
[schema]
(JsonRawSchema/from (slurp (m/encode "application/json" schema))))
(declare of)
(defn- object-schema ^JsonObjectSchema [schema]
(let [builder (JsonObjectSchema/builder)]
(doseq [[k v] (field schema :properties)]
(.addProperty builder (name k) (of v)))
(when-let [required (seq (field schema :required))]
(.required builder ^java.util.List (mapv name required)))
(when-let [d (field schema :description)]
(.description builder d))
(.build builder)))
(defn- array-schema ^JsonArraySchema [schema]
(let [builder (JsonArraySchema/builder)]
(when-let [items (field schema :items)]
(.items builder (of items)))
(when-let [d (field schema :description)]
(.description builder d))
(.build builder)))
(defn- enum-schema ^JsonEnumSchema [schema enum-values]
(let [builder (.enumValues (JsonEnumSchema/builder) ^java.util.List (mapv str enum-values))]
(when-let [d (field schema :description)]
(.description builder d))
(.build builder)))
(defmacro ^:private simple-schema [ctor schema]
`(let [builder# (~ctor)]
(when-let [d# (field ~schema :description)]
(.description builder# d#))
(.build builder#)))
(defn of
"`schema` (a JSON Schema map, keys keyword or string) as a `JsonSchemaElement`."
^JsonSchemaElement [schema]
(if-not (map? schema)
(raw schema)
(let [type (some-> (field schema :type) name str/lower-case)
enum-values (field schema :enum)]
(cond
(and (seq enum-values) (or (nil? type) (= "string" type))) (enum-schema schema enum-values)
(= "object" type) (object-schema schema)
(= "array" type) (array-schema schema)
(= "string" type) (simple-schema #(JsonStringSchema/builder) schema)
(= "integer" type) (simple-schema #(JsonIntegerSchema/builder) schema)
(= "number" type) (simple-schema #(JsonNumberSchema/builder) schema)
(= "boolean" type) (simple-schema #(JsonBooleanSchema/builder) schema)
:else (raw schema)))))
(defn tool-parameters
"A tool's MCP-style `inputSchema` (already validated as an object schema, see tbd.chat.schema/Tool)
as the `JsonObjectSchema` a `ToolSpecification` needs."
^JsonObjectSchema [input-schema]
(object-schema input-schema))
src/tbd/chat/core.clj
-302
View File
@@ -1,302 +0,0 @@
(ns tbd.chat.core
"The chatbot's side of the LLM: a streaming, OpenAI-compatible chat completion request to the LM
Studio server named by the environment. No Ring here; tbd.chat.routes owns that.
`start` sends the request and gives {:ok pump} or {:error text :kind :unavailable|:upstream},
like tbd.todo.service's. `pump` forwards the reply, as it is generated, to a function that takes
events: {:thinking \"text\"} (a piece of the model's reasoning), {:content
\"text\"} (a piece of the reply), {:tool-calls [{:id :name :arguments}]} (the model wants these
tools run, once its reply is complete: `arguments` is the JSON text the model wrote), and
finally {:done true} or {:error \"text\"}.
The chatbot lives on the UI's pages and acts on them, so every request starts with a system
message (`default-system-prompt`, then which page the user is on, `page`, and the tools it has).
The tools are the ones the page offers to the chatbot (see tbd.ui.model-context), described as
MCP does in `tools/list`: {:name :description :inputSchema}. The LLM only sees their metadata: it
asks for a call, the browser makes it and sends the result back with the next request, as a
message with the role \"tool\" (see tbd.chat.schema/Message). This ns translates all of that to
and from the OpenAI-style `tools`, `tool_calls` and `tool_call_id` of the LLM server.
Every turn is traced (see tbd.chat.trace): what was asked, how the reply went, and whether the
model looks stuck in a loop. The config's `:trace` says where that goes."
(:require [clojure.string :as str]
[muuntaja.core :as m]
[tbd.agent.prompt :as prompt]
[tbd.chat.trace :as trace])
(:import (java.io BufferedReader Closeable IOException InputStream InputStreamReader)
(java.net URI)
(java.net.http HttpClient HttpClient$Version HttpRequest HttpRequest$BodyPublishers HttpResponse$BodyHandlers)
(java.nio.charset StandardCharsets)
(java.time Duration)
(java.util.concurrent ScheduledFuture ScheduledThreadPoolExecutor ThreadFactory TimeUnit)))
(def default-timeout-seconds 120)
;; HTTP/1.1 on purpose: the JDK client's default is to ask a plain-http server to upgrade to HTTP/2
;; (`Upgrade: h2c`), and LM Studio's server never answers that request, so the backend would wait
;; out the whole timeout while the LLM sat idle.
(def ^:private http-client
(delay (-> (HttpClient/newBuilder)
(.version HttpClient$Version/HTTP_1_1)
(.connectTimeout (Duration/ofSeconds 5))
.build)))
(defn java-http
"Default `:http` fn: {:url, :headers, :body, :timeout-seconds} is POSTed, giving {:status, :body}
as soon as the response starts; :body is an InputStream of the rest, which the caller closes.
Configs can swap it out, which is how the tests stub the LLM server. It throws when the server
can't be reached or doesn't start answering in time."
[{:keys [url headers body timeout-seconds]}]
(let [builder (-> (HttpRequest/newBuilder (URI/create url))
(.timeout (Duration/ofSeconds timeout-seconds))
(.POST (HttpRequest$BodyPublishers/ofString body)))]
(doseq [[k v] headers] (.header builder k v))
(let [resp (.send ^HttpClient @http-client (.build builder) (HttpResponse$BodyHandlers/ofInputStream))]
{:status (.statusCode resp) :body (.body resp)})))
(defn- env-val [env k]
(let [v (get env k)]
(when-not (str/blank? v) (str/trim v))))
(defn env-config
"Chat config from environment variables:
LMSTUDIO_URL where the LM Studio server is, e.g. http://192.168.1.20:1234 (chat is off without it)
LMSTUDIO_MODEL the model id to ask for (optional: LM Studio uses the model that is loaded)
LMSTUDIO_API_KEY sent as a bearer token, when the server wants one (optional)
LMSTUDIO_TIMEOUT_SECONDS how long the server may stay silent before giving up (default 120)
CHAT_TRACE_FILE a file to append a JSON trace of every turn to, with the model's full reasoning and
replies and the messages it was sent (optional: without it, each turn is only
summarised in the log)"
([] (env-config (System/getenv)))
([env]
{:url (some-> (env-val env "LMSTUDIO_URL") (str/replace #"/+$" ""))
:model (env-val env "LMSTUDIO_MODEL")
:api-key (env-val env "LMSTUDIO_API_KEY")
:timeout-seconds (or (some-> (env-val env "LMSTUDIO_TIMEOUT_SECONDS") parse-long)
default-timeout-seconds)
:trace (trace/config-for (env-val env "CHAT_TRACE_FILE"))
:http java-http}))
(defn configured? [{:keys [url]}]
(some? url))
(defn- fail
"Logs a failure and returns the error text for the client. `text` says nothing about the LLM
server's address or its own message; `detail` (for the log only) does, and is where to look when
chat doesn't work."
([text] (fail text nil))
([text detail]
(println (cond-> (str "[tbd] chat: " text) detail (str " (" detail ")")))
text))
(defn- exception-detail [^Exception e]
(str (class e) ": " (.getMessage e)))
(defn- llm-message
"A message of the conversation (see tbd.chat.schema/Message) as the LLM server has it."
[{:keys [role content tool-calls tool-call-id]}]
(case role
"assistant" (cond-> {:role role :content content}
(seq tool-calls) (assoc :tool_calls
(mapv (fn [{:keys [id name arguments]}]
{:id id :type "function"
:function {:name name :arguments arguments}})
tool-calls)))
"tool" {:role role :tool_call_id tool-call-id :content content}
{:role role :content content}))
(defn- llm-tool
"An MCP tool's metadata as a function the LLM may call."
[{:keys [name description inputSchema]}]
{:type "function"
:function (cond-> {:name name :parameters inputSchema}
description (assoc :description description))})
(def default-system-prompt
"What the LLM is told about its job: an agent embedded in a UI, not a conversationalist. The text
is tbd.agent.prompt/chat, shared with the prompts the page offers to an MCP client outside it
(Claude, Codex, through WebMCP), so that tuning it tunes both."
prompt/chat)
(defn- system-message
"Everything the LLM is told before the conversation, as one message: the instructions
(`default-system-prompt`), then, last, where the user is (`page`) and the names of the `tools`
on offer there. The tools' names are in the request's `tools` too, but a model picks them
more readily when the instructions say them as well, and it can say what it can do without a
call. One message, and the first: several chat templates (Gemma's, Mistral's) refuse more than
one system message, or one that isn't first."
[page tools]
{:role "system"
:content (cond-> default-system-prompt
page (str "\n\nThe user is on the \"" page "\" page of the app: what it shows is what they can see.")
(seq tools) (str (if page " Its" "\n\nYour")
" tools: " (str/join ", " (map :name tools))
". The tools of other pages are offered once you navigate there."))})
(defn- request [{:keys [model api-key timeout-seconds] :as config} messages tools page]
{:url (str (:url config) "/v1/chat/completions")
:headers (cond-> {"Content-Type" "application/json"}
api-key (assoc "Authorization" (str "Bearer " api-key)))
:body (slurp (m/encode "application/json"
(cond-> {:messages (into [(system-message page tools)] (map llm-message) messages)
:stream true}
model (assoc :model model)
(seq tools) (assoc :tools (mapv llm-tool tools)))))
:timeout-seconds timeout-seconds})
;; --- Reading the reply ---
(def ^:private watchdog
(delay (doto (ScheduledThreadPoolExecutor.
1 (reify ThreadFactory
(newThread [_ r] (doto (Thread. ^Runnable r "chat-watchdog") (.setDaemon true)))))
(.setRemoveOnCancelPolicy true))))
(defn- call-deltas
"The pieces of tool calls in a delta: the model writes a call bit by bit, over many events. The
first piece of a call has its id and the tool's name, the rest the JSON of its arguments; `index`
says which call (of the several a reply may ask for) a piece belongs to."
[delta]
(when (sequential? (:tool_calls delta))
(for [{:keys [index id function]} (:tool_calls delta)]
{:index (or index 0) :id id :name (:name function) :arguments (:arguments function)})))
(defn- event-of
"What one line of the LLM's event stream says: a map with :content s and/or :thinking s (a piece
of the reasoning: kept as it is, blanks too, since a piece can be just the line break between two
paragraphs), and :call-deltas (see call-deltas) if there are any; {:error text :detail text}; or
:end (`data: [DONE]`). nil for a line that says nothing: blank, a comment, or an event without
text. A map may also have :finish-reason (why the model stopped: \"stop\", \"tool_calls\",
\"length\" for a reply that ran out of tokens) and :usage ({:prompt_tokens :completion_tokens
:total_tokens}, when the server counts): they are for the trace, not the client."
[line]
(when-let [data (some-> (re-matches #"data:\s?(.*)" line) second str/trim)]
(if (= "[DONE]" data)
:end
(let [event (try (m/decode "application/json" data) (catch Exception _ nil))
delta (get-in event [:choices 0 :delta])
text? (not (str/blank? (:content delta)))
thinking (:reasoning_content delta)
finish-reason (get-in event [:choices 0 :finish_reason])
usage (:usage event)]
(if (:error event)
{:error "the LLM server reported an error" :detail (pr-str (:error event))}
(not-empty
(cond-> {}
text? (assoc :content (:content delta))
(and (string? thinking) (seq thinking)) (assoc :thinking thinking)
(seq (call-deltas delta)) (assoc :call-deltas (vec (call-deltas delta)))
(string? finish-reason) (assoc :finish-reason finish-reason)
(map? usage) (assoc :usage (select-keys usage [:prompt_tokens :completion_tokens :total_tokens])))))))))
(defn- add-deltas
"`calls` ({index {:id :name :arguments}}, in order) with the pieces of calls in `deltas`."
[calls deltas]
(reduce (fn [calls {:keys [index id arguments] :as delta}]
(update calls index
(fn [call]
(cond-> (or call {:arguments ""})
id (assoc :id id)
(:name delta) (assoc :name (:name delta))
arguments (update :arguments str arguments)))))
calls
deltas))
(defn- finished-calls
"The calls the model asked for, as {:id :name :arguments}. A call that never got a tool's name
is no call."
[calls]
(vec (for [[index {:keys [id arguments] :as call}] calls
:when (not (str/blank? (:name call)))]
{:id (or id (str "call_" index)) :name (:name call) :arguments arguments})))
(defn- pump
"Reads the LLM's event stream `in`, calling `emit` with each event (see the ns docstring), and
closes `in`. More than `timeout-seconds` of silence from the LLM is an error. An exception from
`emit` (the client went away, say) ends it too, and is thrown to the caller. `tracer` (see
tbd.chat.trace) sees everything the LLM says, and how the turn ended, before the client is told."
[^InputStream in timeout-seconds emit tracer]
(let [timer (atom nil)
timed-out? (atom false)
arm! (fn []
(some-> ^ScheduledFuture @timer (.cancel false))
(reset! timer (.schedule ^ScheduledThreadPoolExecutor @watchdog
^Runnable (fn []
(reset! timed-out? true)
(try (.close in) (catch IOException _ nil)))
(long timeout-seconds) TimeUnit/SECONDS)))]
(arm!)
(try
(with-open [reader (BufferedReader. (InputStreamReader. in StandardCharsets/UTF_8))]
(loop [replied? false
calls (sorted-map)]
(let [line (try (.readLine reader)
(catch IOException e
(if @timed-out?
{:error "the LLM server took too long to answer"
:detail (str timeout-seconds "s of silence")}
{:error "the LLM server's reply was cut off" :detail (exception-detail e)})))
event (cond (map? line) line ; the read failed
(nil? line) :end ; the stream ended without [DONE]
:else (event-of line))]
(cond
(nil? event) (do (arm!) (recur replied? calls))
(= :end event) (let [calls (finished-calls calls)]
(cond
(seq calls) (do (trace/finish! tracer {:outcome :tool-calls :calls calls})
(emit {:tool-calls calls})
(emit {:done true}))
replied? (do (trace/finish! tracer {:outcome :done})
(emit {:done true}))
:else (let [text (fail "the LLM server sent no reply")]
(trace/finish! tracer {:outcome :error :error text})
(emit {:error text}))))
(:error event) (let [text (fail (:error event) (:detail event))]
(trace/finish! tracer {:outcome :error :error text :detail (:detail event)})
(emit {:error text}))
:else (do (arm!)
(trace/observe! tracer event)
(when-let [out (not-empty (select-keys event [:content :thinking]))] (emit out))
(recur (or replied? (contains? event :content))
(add-deltas calls (:call-deltas event))))))))
(finally
;; Only does anything if the turn ended some other way than the ones above: `emit` threw
;; (the client went away), or something else did.
(trace/finish! tracer {:outcome :aborted})
(some-> ^ScheduledFuture @timer (.cancel false))
(try (.close in) (catch IOException _ nil))))))
(defn start
"Sends the LLM the conversation `messages` (see tbd.chat.schema/Message), and the `tools` it may
ask the page to call (MCP's {:name :description :inputSchema}; none if nil). `page`, if there is
one, is the page of the UI the user is on: the LLM is told so, and which tools it has, in the
system message that goes ahead of the conversation. Gives {:error text
:kind :unavailable|:upstream} when chat isn't configured, or the LLM server can't be reached or
refuses; nothing is streamed then. Otherwise {:ok pump}: `(pump emit)` blocks until the reply is
complete, calling `emit` with its events (see the ns docstring). If `emit` throws, the request to
the LLM is closed (which stops it generating) and the exception is rethrown."
([config messages] (start config messages nil nil))
([config messages tools] (start config messages tools nil))
([{:keys [timeout-seconds http trace] :as config} messages tools page]
(if-not (configured? config)
{:error "chat is not configured" :kind :unavailable}
(let [tracer (trace/begin! trace {:messages messages :tools tools :page page :model (:model config)})
upstream-error (fn [text detail]
(fail text detail)
(trace/finish! tracer {:outcome :upstream-error :error text :detail detail})
{:error text :kind :upstream})]
(try
(let [{:keys [status body]} (http (request config messages tools page))]
(if (= 200 status)
{:ok (fn [emit] (pump body timeout-seconds emit tracer))}
(let [snippet (try (String. (.readNBytes ^InputStream body 200) StandardCharsets/UTF_8)
(catch Exception _ nil))]
(some-> ^Closeable body .close)
(upstream-error (str "the LLM server answered HTTP " status) (str "body: " snippet)))))
(catch java.net.http.HttpTimeoutException e
(upstream-error "the LLM server took too long to answer" (.getMessage e)))
(catch IOException e
(upstream-error "the LLM server can't be reached" (exception-detail e)))
(catch Exception e
(upstream-error "the LLM server sent an unreadable reply" (exception-detail e))))))))
src/tbd/chat/routes.clj
+104 -19
View File
@@ -1,7 +1,18 @@
(ns tbd.chat.routes
"/api/chat: an AG-UI (Agent-User Interaction Protocol, https://docs.ag-ui.com) run. A caller
POSTs a RunAgentInput (tbd.chat.schema/RunAgentInput) and gets the agent's reply streamed back
as AG-UI's server-sent events: RUN_STARTED, then TEXT_MESSAGE_*/REASONING_MESSAGE_*/TOOL_CALL_*
as the model generates, then RUN_FINISHED or RUN_ERROR. The agent itself (its system prompt,
the LLM call, tool-call assembly) is tbd.agent.core; this ns only translates its internal event
shape ({:thinking}/{:content}/{:tool-calls}/{:done}/{:error}, see tbd.agent.core's ns docstring)
to and from AG-UI's wire format.
AG-UI's frontend-tool model — the agent asks for a call, the browser (tbd.ui.webmcp) runs it and
sends the result back as a `tool` message on the next run — is exactly the loop tbd.ui.events
already drove against the old bespoke SSE shape; nothing about that loop changes here."
(:require [muuntaja.core :as m]
[ring.core.protocols :as ring-protocols]
[tbd.chat.core :as chat]
[tbd.agent.core :as agent]
[tbd.chat.schema :as schema]
[tbd.http :refer [respond]])
(:import (java.io IOException OutputStream)
@@ -10,28 +21,102 @@
(def ^:private error-status {:unavailable 503 :upstream 502})
(defn- sse-event
"One server-sent event: its data is `event` as JSON, on a single line."
"One server-sent event: `event` (an AG-UI BaseEvent, a plain map with a `:type`) as JSON, on a
single line."
^bytes [event]
(.getBytes (str "data: " (slurp (m/encode "application/json" event)) "\n\n") StandardCharsets/UTF_8))
;; --- Internal pump events -> AG-UI events ---
(defn- open-reasoning [{:keys [reasoning-id] :as state}]
(if (:reasoning-open? state)
[state []]
[(assoc state :reasoning-open? true)
[{:type "REASONING_START" :messageId reasoning-id}
{:type "REASONING_MESSAGE_START" :messageId reasoning-id :role "reasoning"}]]))
(defn- close-reasoning [{:keys [reasoning-id] :as state}]
(if-not (:reasoning-open? state)
[state []]
[(assoc state :reasoning-open? false)
[{:type "REASONING_MESSAGE_END" :messageId reasoning-id}
{:type "REASONING_END" :messageId reasoning-id}]]))
(defn- open-text [{:keys [text-id] :as state}]
(if (:text-open? state)
[state []]
[(assoc state :text-open? true)
[{:type "TEXT_MESSAGE_START" :messageId text-id :role "assistant"}]]))
(defn- close-text [{:keys [text-id] :as state}]
(if-not (:text-open? state)
[state []]
[(assoc state :text-open? false)
[{:type "TEXT_MESSAGE_END" :messageId text-id}]]))
(defn- close-all [state]
(let [[state text-events] (close-text state)
[state reasoning-events] (close-reasoning state)]
[state (into text-events reasoning-events)]))
(defn- tool-call-events [{:keys [id name arguments]}]
[{:type "TOOL_CALL_START" :toolCallId id :toolCallName name}
{:type "TOOL_CALL_ARGS" :toolCallId id :delta (str arguments)}
{:type "TOOL_CALL_END" :toolCallId id}])
(defn- ag-ui-events
"[state' events]: `state` ({:reasoning-id :text-id :reasoning-open? :text-open?}) and the AG-UI
events `event` (tbd.agent.core's internal shape) translates to, opening/closing the
reasoning/text message streams as the model switches between them."
[state {:keys [thinking content tool-calls done error]} thread-id run-id]
(cond
thinking (let [[state closed] (close-text state)
[state opened] (open-reasoning state)]
[state (conj (into closed opened)
{:type "REASONING_MESSAGE_CONTENT" :messageId (:reasoning-id state) :delta thinking})])
content (let [[state closed] (close-reasoning state)
[state opened] (open-text state)]
[state (conj (into closed opened)
{:type "TEXT_MESSAGE_CONTENT" :messageId (:text-id state) :delta content})])
tool-calls (let [[state closing] (close-all state)]
[state (into closing (mapcat tool-call-events) tool-calls)])
done (let [[state closing] (close-all state)]
[state (conj closing {:type "RUN_FINISHED" :threadId thread-id :runId run-id})])
error (let [[state closing] (close-all state)]
[state (conj closing {:type "RUN_ERROR" :message error})])
:else [state []]))
(defn- event-stream
"A response body that writes what `pump` (see tbd.chat.core/start) forwards, one server-sent
event each, flushing after every one so they arrive as they are generated. A browser that goes
away stops it, and the LLM's request with it."
[pump]
"A response body that writes what `pump` (see tbd.agent.core/start) forwards, translated to AG-UI
events, flushing after every one so they arrive as they are generated. A browser that goes away
stops it, and the LLM's request with it."
[pump thread-id run-id]
(reify ring-protocols/StreamableResponseBody
(write-body-to-stream [_ _ out]
(try
(pump (fn [event]
(.write ^OutputStream out (sse-event event))
(.flush ^OutputStream out)))
(catch IOException _ nil)))))
(let [state (atom {:reasoning-id (str (random-uuid)) :text-id (str (random-uuid))
:reasoning-open? false :text-open? false})
write! (fn [event] (.write ^OutputStream out (sse-event event)) (.flush ^OutputStream out))]
(try
(write! {:type "RUN_STARTED" :threadId thread-id :runId run-id})
(pump (fn [internal-event]
(let [[state' events] (ag-ui-events @state internal-event thread-id run-id)]
(reset! state state')
(run! write! events))))
(catch IOException _ nil))))))
(defn- chat-handler [config]
(fn [{{{:keys [messages tools page]} :body} :parameters}]
;; The LLM is asked before the response starts, so an LLM that is down is still a 502, not an
;; error inside a stream that has already begun.
(let [{:keys [ok error kind]} (chat/start config messages tools page)]
(fn [{{{:keys [messages tools threadId runId] :as input} :body} :parameters}]
(let [thread-id (or threadId (str (random-uuid)))
run-id (or runId (str (random-uuid)))
system-text (schema/system-text input)
internal-messages (mapv schema/->internal-message messages)
internal-tools (mapv schema/->internal-tool tools)
{:keys [ok error kind]} (agent/start config system-text internal-messages internal-tools)]
(if error
(respond (error-status kind) {:error error})
{:status 200
@@ -39,15 +124,15 @@
"Cache-Control" "no-cache"
;; nginx would otherwise hold the events back until the reply is over
"X-Accel-Buffering" "no"}
:body (event-stream ok)}))))
:body (event-stream ok thread-id run-id)}))))
(defn routes
"`config` is the chat config (see tbd.chat.core/env-config). Only logged-in users may chat:
"`config` is the agent config (see tbd.agent.core/env-config). Only logged-in users may chat:
every message costs the LLM server's time."
[config]
["/api/chat"
{:post {:tags ["chat"]
:login? true
:summary "Send a conversation to the LLM (LM Studio); the reply streams back as server-sent events: data: {\"thinking\":\"...\"} for each piece of its reasoning, data: {\"content\":\"...\"} for each piece of the reply, data: {\"tool-calls\":[{\"id\":\"...\",\"name\":\"...\",\"arguments\":\"{...}\"}]} if the model asks for tools of the page (the `tools` in the body, as MCP describes them) to be called, and last data: {\"done\":true} or data: {\"error\":\"...\"}"
:parameters {:body schema/Chat}
:summary "An AG-UI (https://docs.ag-ui.com) run: POST a RunAgentInput (the conversation so far, tools on offer, and `context` — the page the user is on travels as {\"description\":\"page\",\"value\":\"...\"}); the reply streams back as AG-UI server-sent events (RUN_STARTED, TEXT_MESSAGE_*/REASONING_MESSAGE_*/TOOL_CALL_* as the model generates, then RUN_FINISHED or RUN_ERROR). Failures before the run starts are ordinary errors: 400 for an invalid input, 502 if the LLM server can't be reached or refuses, 503 if the agent isn't configured"
:parameters {:body schema/RunAgentInput}
:handler (chat-handler config)}}])
src/tbd/chat/schema.clj
+80 -30
View File
@@ -1,11 +1,15 @@
(ns tbd.chat.schema
"The shape of the chat request, as Malli schemas, coerced by tbd.chat.routes. The limits are
what keeps the endpoint from being a free pipe to the LLM: a conversation is bounded in
length, and the UI (tbd.ui.events) sends no more than `max-messages` of it.
"The shape of /api/chat's request, as Malli schemas, coerced by tbd.chat.routes: an AG-UI
RunAgentInput (https://docs.ag-ui.com), the standard wire format for an agent↔frontend run. The
limits are what keeps the endpoint from being a free pipe to the LLM: a conversation is bounded
in length, and the UI (tbd.ui.events) sends no more than `max-messages` of it.
The conversation can hold the model's use of the page's tools (see tbd.ui.model-context): an
assistant message that asks for calls (`:tool-calls`), followed by one \"tool\" message with the
result of each. Those are the browser's to say, so they are as bounded as the rest.")
assistant message that asks for calls (`:toolCalls`), followed by one \"tool\" role message with
the result of each. Those are the browser's to say, so they are as bounded as the rest. Which
page the user is on travels as a `context` entry ({:description \"page\" :value \"todos\"}) rather
than a dedicated field — AG-UI's place for ambient, non-conversation information."
(:require [tbd.agent.instructions :as instructions]))
(def max-messages 40)
(def max-message-length 8000)
@@ -21,54 +25,100 @@
(<= (count (pr-str tool)) max-tool-schema-length))
(def Tool
"What the page says about one of its tools, as MCP's `tools/list` does. `inputSchema` is JSON Schema
for the arguments, which is the LLM's to read, so it is any (small) object schema."
"AG-UI's Tool: {name, description, parameters}. `parameters` is JSON Schema for the arguments,
which is the LLM's to read, so it is any (small) object."
[:and
[:map
[:name [:re #"^[a-zA-Z0-9_-]{1,64}$"]]
[:description {:optional true} [:string {:max 2000}]]
;; A map-of, not a :map with a :type: the coercion strips the keys a :map doesn't name, and
;; these are all the LLM's to read (`properties`, `required`, ...).
[:inputSchema [:and [:map-of :keyword :any] [:fn (fn [schema] (= "object" (:type schema)))]]]]
[:parameters {:optional true} [:map-of :keyword :any]]]
[:fn small?]])
(def ^:private Id [:string {:min 1 :max 200}])
(def ToolCall
"A call the model asked for. Its name and arguments are the model's own words: nothing says they
are a tool of the page, or JSON, so they are only bounded, generously (a conversation that the
server refused would be stuck: the UI sends it again with the next message)."
(def FunctionCall
[:map
[:id Id]
[:name [:string {:min 1 :max 200}]]
[:arguments [:string {:max 100000}]]])
(def ToolCall
"AG-UI's ToolCall: a call the model asked for. Its name and arguments are the model's own
words: nothing says they are a tool of the page, or JSON, so they are only bounded, generously
(a conversation that the server refused would be stuck: the UI sends it again with the next
message)."
[:map
[:id Id]
[:type [:= "function"]]
[:function FunctionCall]])
(def Message
"AG-UI's Message, narrowed to the roles this endpoint accepts from a caller: user, assistant
(optionally with `toolCalls`) and tool (answering one). `id` identifies the message within the
conversation (AG-UI requires it); this endpoint doesn't need it to be unique or stable, only
present."
[:multi {:dispatch :role}
["user" [:map
[:id Id]
[:role [:= "user"]]
[:content [:string {:min 1 :max max-message-length}]]]]
["assistant" [:and
[:map
[:id Id]
[:role [:= "assistant"]]
[:content [:string {:max max-message-length}]]
[:tool-calls {:optional true} [:vector {:max max-tool-calls} ToolCall]]]
[:fn (fn [{:keys [content tool-calls]}] (or (seq content) (seq tool-calls)))]]]
[:content {:optional true} [:string {:max max-message-length}]]
[:toolCalls {:optional true} [:vector {:max max-tool-calls} ToolCall]]]
[:fn (fn [{:keys [content toolCalls]}] (or (seq content) (seq toolCalls)))]]]
["tool" [:map
[:id Id]
[:role [:= "tool"]]
[:tool-call-id Id]
[:toolCallId Id]
[:content [:string {:min 1 :max max-message-length}]]]]])
(def Page
"Which page of the UI the user is on: its id, as the UI has it (\"tic-tac-toe\", \"ai-setup\"). The server
doesn't know the pages, but the LLM is told it, so it is a short slug and nothing that reads as
an instruction."
[:re #"^[a-z0-9-]{1,64}$"])
(def Chat
"The conversation so far, oldest first; the LLM's reply follows the last message. `page` is the
page the user is on, and `tools` the ones that page offers to the LLM, for this reply."
(def Context
"AG-UI's Context: a named piece of ambient information for the run. The page the user is on is
sent as {:description \"page\" :value \"<page id>\"}."
[:map
[:description [:string {:max 200}]]
[:value [:string {:max 2000}]]])
(def RunAgentInput
"AG-UI's RunAgentInput: the conversation so far, oldest first (`messages`), the tools on offer
for this run (`tools`), and ambient information (`context`, which is where `page` travels).
`threadId`/`runId` identify the run to the client; this server is stateless between turns (the
browser resends the whole conversation each time, see tbd.ui.events), so it only echoes them
back, never reads them."
[:map
[:threadId {:optional true} [:string {:max 200}]]
[:runId {:optional true} [:string {:max 200}]]
[:messages [:vector {:min 1 :max max-messages} Message]]
[:page {:optional true} Page]
[:tools {:optional true} [:vector {:max max-tools} Tool]]])
[:tools {:optional true} [:vector {:max max-tools} Tool]]
[:context {:optional true} [:vector {:max 8} Context]]])
(defn page-of
"The `page` context entry's value, or nil."
[context]
(some #(when (= "page" (:description %)) (:value %)) context))
(defn ->internal-tool
"An AG-UI Tool as the MCP-style {:name :description :inputSchema} tbd.agent.core and
tbd.agent.json-schema expect."
[{:keys [name description parameters]}]
(cond-> {:name name :inputSchema (or parameters {})}
description (assoc :description description)))
(defn ->internal-message
"An AG-UI Message as tbd.agent.core's internal {:role :content :tool-calls :tool-call-id} (see
tbd.agent.core/llm-message)."
[{:keys [role content toolCalls toolCallId]}]
(cond-> {:role role :content content}
(seq toolCalls) (assoc :tool-calls
(mapv (fn [{:keys [id function]}]
{:id id :name (:name function) :arguments (:arguments function)})
toolCalls))
toolCallId (assoc :tool-call-id toolCallId)))
(defn system-text
"The agent's own system prompt for this run (see tbd.agent.instructions), never the caller's to
supply: only `page` (from `context`) and `tools` shape it."
[{:keys [tools context]}]
(instructions/chat-system-message (page-of context) tools))
src/tbd/chat/trace.clj
+10 -7
View File
@@ -1,8 +1,10 @@
(ns tbd.chat.trace
"Observability for the chatbot's LLM turns: what the model was asked, what it did with it, how long
it took, and whether it looks stuck. No Ring and no LLM here; tbd.chat.core calls it.
it took, and whether it looks stuck. No Ring and no LLM here; tbd.agent.core (the chatbot's agent)
and tbd.agent.a2a (the A2A server) both call it.
One request to /api/chat is one turn of the model. The loop that runs tools lives in the browser
One request to /api/chat, or one message to the A2A server, is one turn of the model. The loop that
runs the chatbot's tools lives in the browser
(tbd.ui.events): a tool's result comes back as a new request holding the conversation so far, so the
server keeps nothing between turns. A stuck agent shows up in two ways, and both are watched:
@@ -23,7 +25,7 @@
if given a file, appends it there as JSON. Only the file has the text of the reasoning and the reply
(`:reasoning`, `:reply`) and the run's messages (`:input`): they are what you read to see why a
model is looping, and they are the user's words, so they are off unless a file is asked for
(CHAT_TRACE_FILE, see tbd.chat.core/env-config)."
(CHAT_TRACE_FILE, see tbd.agent.core/env-config)."
(:require [clojure.java.io :as io]
[clojure.string :as str]
[muuntaja.core :as m])
@@ -133,7 +135,8 @@
long? (conj {:kind :long-reasoning :chars chars :elapsed-ms (- now (:t0 state))}))]))
(defn- advance
"[state' alerts]: `state` after one event of the LLM's reply (tbd.chat.core/event-of's map)."
"[state' alerts]: `state` after one event of the LLM's reply (the map tbd.agent.core/observe! is
called with: any of :content, :thinking, :finish-reason, :usage)."
[state {:keys [content thinking call-deltas finish-reason usage]} now full?]
(let [state (cond-> (-> state
(update :events inc)
@@ -238,9 +241,9 @@
tracer))
(defn observe!
"Takes one event of the LLM's reply (see tbd.chat.core/event-of: any of :content, :thinking,
:call-deltas, :finish-reason, :usage), logging a `:loop-suspect` if the reasoning has gone on for
long, or is repeating itself."
"Takes one event of the LLM's reply (see tbd.agent.core, whose pump calls this: :content, :thinking,
or, at the end of a turn, :finish-reason and :usage), logging a `:loop-suspect` if the reasoning has
gone on for long, or is repeating itself."
[{:keys [state full-text?] :as tracer} event]
(let [[state' alerts] (advance @state event (now-ms) full-text?)]
(reset! state (update state' :flags into (map :kind) alerts))
src/tbd/core.clj
+5 -3
View File
@@ -14,10 +14,11 @@
[reitit.ring.middleware.parameters :as parameters]
[muuntaja.core :as m]
[ring.util.response :as response]
[tbd.agent.a2a :as a2a]
[tbd.agent.core :as agent]
[tbd.auth.core :as auth]
[tbd.auth.middleware :as auth-mw]
[tbd.auth.routes :as auth-routes]
[tbd.chat.core :as chat]
[tbd.chat.routes :as chat-routes]
[tbd.health.routes :as health]
[tbd.http :as http]
@@ -86,6 +87,7 @@
(health/routes)
(math/routes)
(mcp-routes/routes auth ds)
(a2a/routes auth ds chat)
(oauth-routes/routes auth ds)
(auth-routes/routes auth)
(chat-routes/routes chat)
@@ -123,7 +125,7 @@
(def config
{:auth (auth/env-config)
:nrepl (nrepl/env-config)
:chat (chat/env-config)
:chat (agent/env-config)
:db (sqlite/env-config)
:webmcp {:origin-trial-token (some-> (System/getenv "WEBMCP_ORIGIN_TRIAL_TOKEN") str/trim not-empty)}})
@@ -144,7 +146,7 @@
(println "[tbd] WARNING: ADMIN_USERS is empty; nobody can use the admin API"))
(when-not (get-in config [:auth :session-secret])
(println "[tbd] WARNING: SESSION_SECRET is not set; logins won't survive a restart"))
(when-not (chat/configured? (:chat config))
(when-not (agent/configured? (:chat config))
(println "[tbd] LMSTUDIO_URL is not set; the chatbot is unavailable"))
(when-let [file (get-in config [:chat :trace :file])]
(println (str "[tbd] chat: tracing every turn, reasoning and replies included, to " file)))