feat(agent): SSE agent route backed by open-agent-sdk via BoosAPI provider
Adds the Agent Workbench HTTP surface wiring the open-agent-sdk loop into
BoosAPI:
- POST /api/agent/chat (SSE): runs an agentic turn and streams session /
text / thinking / tool_use / tool_result / approval_request / result /
error / done events. The agent's LLM calls are routed back through
BoosAPI's own OpenAI-compatible /v1 using the built-in downstream key
(constructed via the SDK's OpenAIProvider), so any aggregated model works
and usage flows through the normal accounting pipeline.
- POST /api/agent/approve: resolves a pending tool approval by request id.
- GET /api/agent/status: engine info + a session's accounting.
Safety is enforced in the SDK permission callback: every tool first passes
sandbox path validation (no escaping the per-session working dir), then the
risk/mode matrix — writes and command execution require an interactive
approval round-trip (default mode) that blocks the loop until the user
answers or the request times out (300s -> deny). Each session runs in its
own data/agent_sessions/{id}/ sandbox and turn usage is persisted.
Registered in the API router and added open-agent-sdk as a path dependency.
This commit is contained in:
boos
@@ -0,0 +1,455 @@
|
||||
//! Agent Workbench HTTP surface.
|
||||
//!
|
||||
//! A general-purpose, server-side AI agent exposed to the SPA over SSE. The
|
||||
//! agentic loop comes from the `open-agent-sdk` crate; every LLM call is routed
|
||||
//! back through BoosAPI's own OpenAI-compatible `/v1` endpoint using the
|
||||
//! built-in downstream key, so agent usage flows through the same aggregation
|
||||
//! and accounting pipeline as any other consumer (and can use any aggregated
|
||||
//! model).
|
||||
//!
|
||||
//! Routes (all require a Bearer token — admin token, proxy token, or a user JWT):
|
||||
//! - `POST /api/agent/chat` Stream a turn as SSE (text / tool_use /
|
||||
//! tool_result / approval_request / result / error / done).
|
||||
//! - `POST /api/agent/approve` Resolve a pending tool approval.
|
||||
//! - `GET /api/agent/status` Engine status + a session's accounting.
|
||||
//!
|
||||
//! ## Safety
|
||||
//! Each session runs inside an isolated sandbox directory
|
||||
//! (`data/agent_sessions/{id}/`). The SDK permission callback ([`build_can_use_tool`])
|
||||
//! enforces two gates on every tool: (1) path arguments must stay within the
|
||||
//! sandbox, and (2) writes/executes require explicit user approval unless the
|
||||
//! session mode opts out. The default mode requires approval for anything that
|
||||
//! mutates state or runs a command.
|
||||
|
||||
use crate::auth::jwt::validate_jwt;
|
||||
use crate::services::agent_workbench::{
|
||||
approval::{self, ApprovalOutcome},
|
||||
permission::{classify_tool_risk, decide_permission, PermissionAction, PermissionMode},
|
||||
sandbox,
|
||||
session_store::{self, UsageDelta},
|
||||
};
|
||||
use crate::AppState;
|
||||
use axum::{
|
||||
extract::{DefaultBodyLimit, State},
|
||||
http::{HeaderMap, StatusCode},
|
||||
response::sse::{Event, Sse},
|
||||
routing::{get, post},
|
||||
Json, Router,
|
||||
};
|
||||
use futures::stream::Stream;
|
||||
use open_agent_sdk::api::openai::OpenAIProvider;
|
||||
use open_agent_sdk::types::{ContentBlock, PermissionDecision};
|
||||
use open_agent_sdk::{Agent, AgentOptions, LLMProvider, SDKMessage};
|
||||
use serde::Deserialize;
|
||||
use serde_json::{json, Value};
|
||||
use std::convert::Infallible;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
|
||||
use tokio_stream::wrappers::UnboundedReceiverStream;
|
||||
use uuid::Uuid;
|
||||
|
||||
/// Max agentic turns per chat request (bounds runaway loops / cost).
|
||||
const MAX_TURNS: u32 = 20;
|
||||
/// How long a tool approval waits for the user before auto-denying.
|
||||
const APPROVAL_TIMEOUT: Duration = Duration::from_secs(300);
|
||||
/// Body limit for a chat request (images/attachments can be large).
|
||||
const AGENT_CHAT_BODY_LIMIT_BYTES: usize = 16 * 1024 * 1024;
|
||||
|
||||
const SYSTEM_PROMPT: &str = "You are BoosAPI's built-in engineering agent. You can read and write files \
|
||||
and run shell commands, but only inside the current working directory, which is an \
|
||||
isolated sandbox for this session. Be concise. Explain what you are about to do before \
|
||||
using a tool that writes files or runs commands. Never attempt to access paths outside \
|
||||
the working directory.";
|
||||
|
||||
pub fn routes() -> Router<AppState> {
|
||||
Router::new()
|
||||
.route("/api/agent/chat", post(chat_handler))
|
||||
.route("/api/agent/approve", post(approve_handler))
|
||||
.route("/api/agent/status", get(status_handler))
|
||||
.layer(DefaultBodyLimit::max(AGENT_CHAT_BODY_LIMIT_BYTES))
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct ChatRequest {
|
||||
/// The user's prompt for this turn.
|
||||
message: String,
|
||||
/// Existing session id to continue; a new one is minted when absent.
|
||||
#[serde(default)]
|
||||
session_id: Option<String>,
|
||||
/// Aggregated model id; falls back to the configured agent model.
|
||||
#[serde(default)]
|
||||
model: Option<String>,
|
||||
/// Permission mode: "default" | "accept_edits" | "full_access".
|
||||
#[serde(default)]
|
||||
permission_mode: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct ApproveRequest {
|
||||
request_id: String,
|
||||
approved: bool,
|
||||
}
|
||||
|
||||
/// Resolved caller identity for an agent request.
|
||||
#[derive(Debug, Clone)]
|
||||
struct AgentAuth {
|
||||
/// `None` for admin/global-token callers; `Some(id)` for user JWTs.
|
||||
user_id: Option<i64>,
|
||||
}
|
||||
|
||||
fn extract_bearer_token(headers: &HeaderMap) -> Option<&str> {
|
||||
headers
|
||||
.get(axum::http::header::AUTHORIZATION)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.and_then(|v| v.strip_prefix("Bearer "))
|
||||
.map(str::trim)
|
||||
.filter(|t| !t.is_empty())
|
||||
}
|
||||
|
||||
/// Authenticate the caller. Admin token / proxy token map to an admin
|
||||
/// (user-less) session; a valid user/admin JWT maps to that user's id.
|
||||
async fn resolve_auth(
|
||||
headers: &HeaderMap,
|
||||
state: &AppState,
|
||||
) -> Result<AgentAuth, (StatusCode, Json<Value>)> {
|
||||
let unauthorized = || {
|
||||
(
|
||||
StatusCode::UNAUTHORIZED,
|
||||
Json(json!({ "error": "Unauthorized" })),
|
||||
)
|
||||
};
|
||||
|
||||
let token = extract_bearer_token(headers).ok_or_else(unauthorized)?;
|
||||
|
||||
if token == state.config.auth.auth_token || token == state.runtime_proxy_token() {
|
||||
return Ok(AgentAuth { user_id: None });
|
||||
}
|
||||
|
||||
let claims = validate_jwt(&state.config.auth.jwt_secret, token).map_err(|_| unauthorized())?;
|
||||
if claims.role != "admin" && claims.role != "user" {
|
||||
return Err((StatusCode::FORBIDDEN, Json(json!({ "error": "Forbidden" }))));
|
||||
}
|
||||
let user_id = claims
|
||||
.sub
|
||||
.parse::<i64>()
|
||||
.map_err(|_| (StatusCode::UNAUTHORIZED, Json(json!({ "error": "Invalid user id" }))))?;
|
||||
Ok(AgentAuth { user_id: Some(user_id) })
|
||||
}
|
||||
|
||||
/// Send one SSE event; returns false if the receiver (client) has gone away.
|
||||
fn send_sse(tx: &UnboundedSender<Result<Event, Infallible>>, payload: Value) -> bool {
|
||||
tx.send(Ok(Event::default().data(payload.to_string()))).is_ok()
|
||||
}
|
||||
|
||||
/// Resolve BoosAPI's own base URL (no `/v1` suffix — the OpenAI provider appends
|
||||
/// `/v1/chat/completions`).
|
||||
fn boosapi_base(state: &AppState) -> String {
|
||||
let host = match state.config.server.host.as_str() {
|
||||
"0.0.0.0" | "::" => "127.0.0.1",
|
||||
other => other,
|
||||
};
|
||||
format!("http://{}:{}", host, state.config.server.port)
|
||||
}
|
||||
|
||||
/// Build the SDK permission callback that enforces sandbox path safety and the
|
||||
/// approval flow. The callback runs inside the agent loop for every tool call.
|
||||
///
|
||||
/// Order of checks:
|
||||
/// 1. Sandbox: any path argument that escapes `cwd` is hard-denied immediately.
|
||||
/// 2. Risk + mode: read-only (and, per mode, writes) auto-allow; otherwise emit
|
||||
/// an `approval_request` SSE event and block until the user responds (or the
|
||||
/// approval times out, which denies).
|
||||
fn build_can_use_tool(
|
||||
sse_tx: UnboundedSender<Result<Event, Infallible>>,
|
||||
cwd: std::path::PathBuf,
|
||||
mode: PermissionMode,
|
||||
) -> open_agent_sdk::types::CanUseToolFn {
|
||||
Arc::new(move |tool_name: &str, input: &Value| {
|
||||
let sse_tx = sse_tx.clone();
|
||||
let cwd = cwd.clone();
|
||||
let tool_name = tool_name.to_string();
|
||||
let input = input.clone();
|
||||
Box::pin(async move {
|
||||
// Gate 1: sandbox path safety (always enforced, even in full access).
|
||||
if let Err(reason) = sandbox::check_tool_paths(&tool_name, &input, &cwd) {
|
||||
send_sse(
|
||||
&sse_tx,
|
||||
json!({
|
||||
"type": "tool_denied",
|
||||
"data": { "name": tool_name, "reason": reason }
|
||||
}),
|
||||
);
|
||||
return PermissionDecision::Deny(reason);
|
||||
}
|
||||
|
||||
// Gate 2: risk + permission mode.
|
||||
let risk = classify_tool_risk(&tool_name);
|
||||
match decide_permission(mode, risk) {
|
||||
PermissionAction::AutoAllow => PermissionDecision::Allow,
|
||||
PermissionAction::RequireApproval => {
|
||||
let request_id = Uuid::new_v4().to_string();
|
||||
let rx = approval::register(&request_id);
|
||||
send_sse(
|
||||
&sse_tx,
|
||||
json!({
|
||||
"type": "approval_request",
|
||||
"data": {
|
||||
"request_id": request_id,
|
||||
"name": tool_name,
|
||||
"input": input,
|
||||
"risk": format!("{risk:?}"),
|
||||
}
|
||||
}),
|
||||
);
|
||||
|
||||
match tokio::time::timeout(APPROVAL_TIMEOUT, rx).await {
|
||||
Ok(Ok(ApprovalOutcome::Approve)) => PermissionDecision::Allow,
|
||||
Ok(Ok(ApprovalOutcome::Deny)) => {
|
||||
PermissionDecision::Deny("Denied by user".to_string())
|
||||
}
|
||||
// Sender dropped (session ended) or timed out.
|
||||
_ => {
|
||||
approval::cancel(&request_id);
|
||||
PermissionDecision::Deny("Approval timed out".to_string())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
#[allow(clippy::type_complexity)]
|
||||
async fn chat_handler(
|
||||
State(state): State<AppState>,
|
||||
headers: HeaderMap,
|
||||
Json(body): Json<ChatRequest>,
|
||||
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, (StatusCode, Json<Value>)> {
|
||||
let auth = resolve_auth(&headers, &state).await?;
|
||||
|
||||
if body.message.trim().is_empty() {
|
||||
return Err((
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({ "error": "message is required" })),
|
||||
));
|
||||
}
|
||||
|
||||
let session_id = body
|
||||
.session_id
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(ToOwned::to_owned)
|
||||
.unwrap_or_else(|| Uuid::new_v4().to_string());
|
||||
let mode = PermissionMode::from_wire(body.permission_mode.as_deref().unwrap_or("default"));
|
||||
let model = body
|
||||
.model
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(ToOwned::to_owned)
|
||||
.unwrap_or_else(|| state.config.comfyui.agent_model.clone());
|
||||
|
||||
// Per-session sandbox working directory.
|
||||
let cwd = match sandbox::ensure_session_dir(&session_id) {
|
||||
Ok(dir) => dir,
|
||||
Err(e) => {
|
||||
return Err((
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
Json(json!({ "error": format!("failed to create session sandbox: {e}") })),
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
// Persist (idempotently) the session row before streaming.
|
||||
if let Err(e) = session_store::ensure_session(
|
||||
&state.db,
|
||||
&session_id,
|
||||
auth.user_id,
|
||||
&model,
|
||||
&format!("{mode:?}").to_lowercase(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(%e, "failed to ensure agent session row");
|
||||
}
|
||||
|
||||
// Route the agent's LLM calls back through BoosAPI's own aggregation using
|
||||
// the built-in downstream key.
|
||||
let base_url = boosapi_base(&state);
|
||||
let internal_key = state.builtin_keys.comfyui_agent.1.clone();
|
||||
|
||||
let (tx, rx) = unbounded_channel::<Result<Event, Infallible>>();
|
||||
|
||||
let db = state.db.clone();
|
||||
let prompt = body.message;
|
||||
let session_for_event = session_id.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
// First event: hand the (possibly newly minted) session id to the client.
|
||||
send_sse(
|
||||
&tx,
|
||||
json!({ "type": "session", "data": { "session_id": session_for_event, "model": model } }),
|
||||
);
|
||||
|
||||
let provider: Arc<dyn LLMProvider> = Arc::new(OpenAIProvider::new(
|
||||
reqwest::Client::new(),
|
||||
internal_key,
|
||||
base_url,
|
||||
Default::default(),
|
||||
));
|
||||
|
||||
let can_use_tool = build_can_use_tool(tx.clone(), cwd.clone(), mode);
|
||||
|
||||
let agent = Agent::new(AgentOptions {
|
||||
provider: Some(provider),
|
||||
model: Some(model.clone()),
|
||||
cwd: Some(cwd.to_string_lossy().to_string()),
|
||||
system_prompt: Some(SYSTEM_PROMPT.to_string()),
|
||||
max_turns: Some(MAX_TURNS),
|
||||
can_use_tool: Some(can_use_tool),
|
||||
session_id: Some(session_id.clone()),
|
||||
..Default::default()
|
||||
})
|
||||
.await;
|
||||
|
||||
let mut agent = match agent {
|
||||
Ok(a) => a,
|
||||
Err(e) => {
|
||||
send_sse(&tx, json!({ "type": "error", "data": { "message": e } }));
|
||||
send_sse(&tx, json!({ "type": "done" }));
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let (mut rx_msg, handle) = agent.query(&prompt).await;
|
||||
|
||||
while let Some(event) = rx_msg.recv().await {
|
||||
// Stop early if the client disconnected.
|
||||
if tx.is_closed() {
|
||||
break;
|
||||
}
|
||||
forward_sdk_event(&tx, &db, &session_id, event).await;
|
||||
}
|
||||
let _ = handle.await;
|
||||
agent.close().await;
|
||||
|
||||
send_sse(&tx, json!({ "type": "done" }));
|
||||
});
|
||||
|
||||
Ok(Sse::new(UnboundedReceiverStream::new(rx)))
|
||||
}
|
||||
|
||||
/// Translate one SDK message into SSE event(s), and persist usage on Result.
|
||||
async fn forward_sdk_event(
|
||||
tx: &UnboundedSender<Result<Event, Infallible>>,
|
||||
db: &sqlx::SqlitePool,
|
||||
session_id: &str,
|
||||
event: SDKMessage,
|
||||
) {
|
||||
match event {
|
||||
SDKMessage::Assistant { message, .. } => {
|
||||
for block in &message.content {
|
||||
match block {
|
||||
ContentBlock::Text { text } if !text.is_empty() => {
|
||||
send_sse(tx, json!({ "type": "text", "data": { "text": text } }));
|
||||
}
|
||||
ContentBlock::ToolUse { id, name, input } => {
|
||||
send_sse(
|
||||
tx,
|
||||
json!({ "type": "tool_use", "data": { "id": id, "name": name, "input": input } }),
|
||||
);
|
||||
}
|
||||
ContentBlock::Thinking { thinking, .. } if !thinking.is_empty() => {
|
||||
send_sse(tx, json!({ "type": "thinking", "data": { "thinking": thinking } }));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
SDKMessage::ToolResult { tool_use_id, tool_name, content, is_error } => {
|
||||
send_sse(
|
||||
tx,
|
||||
json!({
|
||||
"type": "tool_result",
|
||||
"data": { "tool_use_id": tool_use_id, "name": tool_name, "content": content, "is_error": is_error }
|
||||
}),
|
||||
);
|
||||
}
|
||||
SDKMessage::Result { text, usage, num_turns, cost_usd, .. } => {
|
||||
// Persist this turn's accounting (user prompt + assistant reply ≈ 2 msgs).
|
||||
let delta = UsageDelta {
|
||||
input_tokens: usage.input_tokens as i64,
|
||||
output_tokens: usage.output_tokens as i64,
|
||||
cost_usd,
|
||||
num_turns: num_turns as i64,
|
||||
};
|
||||
if let Err(e) = session_store::record_turn(db, session_id, delta, 2).await {
|
||||
tracing::warn!(%e, "failed to record agent turn usage");
|
||||
}
|
||||
send_sse(
|
||||
tx,
|
||||
json!({
|
||||
"type": "result",
|
||||
"data": {
|
||||
"text": text,
|
||||
"num_turns": num_turns,
|
||||
"input_tokens": usage.input_tokens,
|
||||
"output_tokens": usage.output_tokens,
|
||||
"total_tokens": usage.input_tokens + usage.output_tokens,
|
||||
"cost_usd": cost_usd,
|
||||
}
|
||||
}),
|
||||
);
|
||||
}
|
||||
SDKMessage::Error { message } => {
|
||||
send_sse(tx, json!({ "type": "error", "data": { "message": message } }));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
async fn approve_handler(
|
||||
State(state): State<AppState>,
|
||||
headers: HeaderMap,
|
||||
Json(body): Json<ApproveRequest>,
|
||||
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
|
||||
// Auth required, but any authenticated caller may answer the prompt for
|
||||
// their own running session (approvals are keyed by an unguessable uuid).
|
||||
resolve_auth(&headers, &state).await?;
|
||||
let outcome = if body.approved { ApprovalOutcome::Approve } else { ApprovalOutcome::Deny };
|
||||
let delivered = approval::resolve(&body.request_id, outcome);
|
||||
Ok(Json(json!({ "ok": delivered })))
|
||||
}
|
||||
|
||||
async fn status_handler(
|
||||
State(state): State<AppState>,
|
||||
headers: HeaderMap,
|
||||
axum::extract::Query(q): axum::extract::Query<StatusQuery>,
|
||||
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
|
||||
let auth = resolve_auth(&headers, &state).await?;
|
||||
let session = match q.session_id.as_deref().filter(|s| !s.is_empty()) {
|
||||
Some(sid) => session_store::get_session(&state.db, sid, auth.user_id)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
(
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
Json(json!({ "error": e.to_string() })),
|
||||
)
|
||||
})?,
|
||||
None => None,
|
||||
};
|
||||
Ok(Json(json!({
|
||||
"ok": true,
|
||||
"engine": "open-agent-sdk",
|
||||
"default_model": state.config.comfyui.agent_model,
|
||||
"session": session,
|
||||
})))
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct StatusQuery {
|
||||
#[serde(default)]
|
||||
session_id: Option<String>,
|
||||
}
|
||||
@@ -22,6 +22,7 @@ const _: &[u8] = include_bytes!(concat!(
|
||||
));
|
||||
|
||||
mod admin;
|
||||
mod agent;
|
||||
mod auth;
|
||||
mod checkin;
|
||||
mod comfy_cloud_adapter;
|
||||
@@ -57,6 +58,7 @@ pub fn build_router(state: AppState) -> Router {
|
||||
.merge(comfy_cloud_adapter::routes())
|
||||
.merge(money_printer::routes())
|
||||
.merge(comfyui_agent::routes())
|
||||
.merge(agent::routes())
|
||||
.merge(digital_human::routes())
|
||||
.merge(comfyui::routes())
|
||||
.merge(crate::workflow::routes::routes())
|
||||
|
||||
Reference in New Issue
Block a user