1use anyhow::Result;
2use axum::extract::DefaultBodyLimit;
3use axum::routing::get;
4use axum::{Json, Router};
5use serde_json::json;
6use systemprompt_database::DatabaseQuery;
7use systemprompt_models::api::SingleResponse;
8use systemprompt_models::modules::ApiPaths;
9use systemprompt_models::AppPaths;
10use systemprompt_runtime::AppContext;
11use systemprompt_traits::{StartupEvent, StartupEventExt, StartupEventSender};
12
13use super::routes::configure_routes;
14use crate::models::ServerConfig;
15use crate::services::middleware::{
16 inject_trace_header, remove_trailing_slash, AnalyticsMiddleware, ContextMiddleware,
17 CorsMiddleware, JwtContextExtractor, SessionMiddleware,
18};
19
20const HEALTH_CHECK_QUERY: DatabaseQuery = DatabaseQuery::new("SELECT 1");
21
22#[derive(Debug)]
23pub struct ApiServer {
24 router: Router,
25 _config: ServerConfig,
26 events: Option<StartupEventSender>,
27}
28
29impl ApiServer {
30 pub fn new(router: Router, events: Option<StartupEventSender>) -> Self {
31 Self::with_config(router, ServerConfig::default(), events)
32 }
33
34 pub const fn with_config(
35 router: Router,
36 config: ServerConfig,
37 events: Option<StartupEventSender>,
38 ) -> Self {
39 Self {
40 router,
41 _config: config,
42 events,
43 }
44 }
45
46 pub async fn serve(self, addr: &str) -> Result<()> {
47 if let Some(ref tx) = self.events {
48 if tx
49 .unbounded_send(StartupEvent::ServerBinding {
50 address: addr.to_string(),
51 })
52 .is_err()
53 {
54 tracing::debug!("Startup event receiver dropped");
55 }
56 }
57
58 let listener = self.create_listener(addr).await?;
59
60 if let Some(ref tx) = self.events {
61 tx.server_listening(addr, std::process::id());
62 }
63
64 axum::serve(
65 listener,
66 self.router
67 .into_make_service_with_connect_info::<std::net::SocketAddr>(),
68 )
69 .await?;
70 Ok(())
71 }
72
73 async fn create_listener(&self, addr: &str) -> Result<tokio::net::TcpListener> {
74 tokio::net::TcpListener::bind(addr)
75 .await
76 .map_err(|e| anyhow::anyhow!("Failed to bind to {addr}: {e}"))
77 }
78}
79
80pub fn setup_api_server(ctx: &AppContext, events: Option<StartupEventSender>) -> Result<ApiServer> {
81 let rate_config = &ctx.config().rate_limits;
82
83 if rate_config.disabled {
84 if let Some(ref tx) = events {
85 tx.warning("Rate limiting disabled - development mode only");
86 }
87 }
88
89 let router = configure_routes(ctx, events.as_ref())?;
90 let router = apply_global_middleware(router, ctx)?;
91
92 Ok(ApiServer::new(router, events))
93}
94
95fn apply_global_middleware(router: Router, ctx: &AppContext) -> Result<Router> {
96 let mut router = router;
97
98 router = router.layer(DefaultBodyLimit::max(100 * 1024 * 1024));
99
100 let analytics_middleware = AnalyticsMiddleware::new(ctx);
101 router = router.layer(axum::middleware::from_fn({
102 let middleware = analytics_middleware;
103 move |req, next| {
104 let middleware = middleware.clone();
105 async move { middleware.track_request(req, next).await }
106 }
107 }));
108
109 let jwt_extractor = JwtContextExtractor::new(
110 systemprompt_models::SecretsBootstrap::jwt_secret()?,
111 ctx.db_pool(),
112 );
113 let global_context_middleware = ContextMiddleware::public(jwt_extractor);
114 router = router.layer(axum::middleware::from_fn({
115 let middleware = global_context_middleware;
116 move |req, next| {
117 let middleware = middleware.clone();
118 async move { middleware.handle(req, next).await }
119 }
120 }));
121
122 let session_middleware = SessionMiddleware::new(ctx)?;
123 router = router.layer(axum::middleware::from_fn({
124 let middleware = session_middleware;
125 move |req, next| {
126 let middleware = middleware.clone();
127 async move { middleware.handle(req, next).await }
128 }
129 }));
130
131 let cors = CorsMiddleware::build_layer(ctx.config())?;
132 router = router.layer(cors);
133
134 router = router.layer(axum::middleware::from_fn(remove_trailing_slash));
135
136 router = router.layer(axum::middleware::from_fn(inject_trace_header));
137
138 Ok(router)
139}
140
141pub async fn handle_root_discovery(
142 axum::extract::State(ctx): axum::extract::State<AppContext>,
143) -> impl axum::response::IntoResponse {
144 let base = &ctx.config().api_external_url;
145 let data = json!({
146 "name": format!("{} API", ctx.config().sitename),
147 "version": "1.0.0",
148 "description": "systemprompt.io OS API Gateway",
149 "endpoints": {
150 "health": format!("{}{}", base, ApiPaths::HEALTH),
151 "oauth": {
152 "href": format!("{}{}", base, ApiPaths::OAUTH_BASE),
153 "description": "OAuth2/OIDC authentication and WebAuthn",
154 "endpoints": {
155 "authorize": format!("{}{}", base, ApiPaths::OAUTH_AUTHORIZE),
156 "token": format!("{}{}", base, ApiPaths::OAUTH_TOKEN),
157 "userinfo": format!("{}{}/userinfo", base, ApiPaths::OAUTH_BASE),
158 "introspect": format!("{}{}/introspect", base, ApiPaths::OAUTH_BASE),
159 "revoke": format!("{}{}/revoke", base, ApiPaths::OAUTH_BASE),
160 "webauthn": format!("{}{}/webauthn", base, ApiPaths::OAUTH_BASE)
161 }
162 },
163 "core": {
164 "href": format!("{}{}", base, ApiPaths::CORE_BASE),
165 "description": "Core conversation, task, and artifact management",
166 "endpoints": {
167 "contexts": format!("{}{}", base, ApiPaths::CORE_CONTEXTS),
168 "tasks": format!("{}{}", base, ApiPaths::CORE_TASKS),
169 "artifacts": format!("{}{}", base, ApiPaths::CORE_ARTIFACTS)
170 }
171 },
172 "agents": {
173 "href": format!("{}{}", base, ApiPaths::AGENTS_REGISTRY),
174 "description": "A2A protocol agent registry and proxy",
175 "endpoints": {
176 "registry": format!("{}{}", base, ApiPaths::AGENTS_REGISTRY),
177 "proxy": format!("{}{}{{agent_id}}", base, ApiPaths::AGENTS_BASE)
178 }
179 },
180 "mcp": {
181 "href": format!("{}{}", base, ApiPaths::MCP_REGISTRY),
182 "description": "MCP server registry and lifecycle management",
183 "endpoints": {
184 "registry": format!("{}{}", base, ApiPaths::MCP_REGISTRY),
185 "proxy": format!("{}{}{{server_name}}", base, ApiPaths::MCP_BASE)
186 }
187 },
188 "stream": {
189 "href": format!("{}{}", base, ApiPaths::STREAM_BASE),
190 "description": "Server-Sent Events (SSE) for real-time updates",
191 "endpoints": {
192 "contexts": format!("{}{}", base, ApiPaths::STREAM_CONTEXTS)
193 }
194 }
195 },
196 "wellknown": {
197 "oauth": format!("{}{}", base, ApiPaths::WELLKNOWN_OAUTH_SERVER),
198 "agent": format!("{}{}", base, ApiPaths::WELLKNOWN_AGENT_CARD)
199 }
200 });
201
202 Json(SingleResponse::new(data))
203}
204
205fn parse_proc_status_kb(content: &str, key: &str) -> Option<u64> {
206 content
207 .lines()
208 .find(|line| line.starts_with(key))
209 .and_then(|line| {
210 line.split_whitespace()
211 .nth(1)
212 .and_then(|v| v.parse::<u64>().ok())
213 })
214}
215
216fn get_process_memory() -> Option<serde_json::Value> {
217 let content = std::fs::read_to_string("/proc/self/status").ok()?;
218
219 let rss_kb = parse_proc_status_kb(&content, "VmRSS:");
220 let virt_kb = parse_proc_status_kb(&content, "VmSize:");
221 let peak_kb = parse_proc_status_kb(&content, "VmPeak:");
222
223 Some(json!({
224 "rss_mb": rss_kb.map(|kb| kb / 1024),
225 "virtual_mb": virt_kb.map(|kb| kb / 1024),
226 "peak_mb": peak_kb.map(|kb| kb / 1024)
227 }))
228}
229
230pub async fn handle_health(
231 axum::extract::State(ctx): axum::extract::State<AppContext>,
232) -> impl axum::response::IntoResponse {
233 use axum::http::StatusCode;
234 use systemprompt_database::{DatabaseProvider, ServiceRepository};
235
236 let start = std::time::Instant::now();
237
238 let (db_status, db_latency_ms) = {
239 let db_start = std::time::Instant::now();
240 let status = match ctx.db_pool().fetch_optional(&HEALTH_CHECK_QUERY, &[]).await {
241 Ok(_) => "healthy",
242 Err(_) => "unhealthy",
243 };
244 (status, db_start.elapsed().as_millis())
245 };
246
247 let service_repo = ServiceRepository::new(ctx.db_pool().clone());
248
249 let (agent_count, agent_status) = match service_repo.count_running_services("agent").await {
250 Ok(count) if count > 0 => (count, "healthy"),
251 Ok(_) => (0, "none"),
252 Err(_) => (0, "error"),
253 };
254
255 let (mcp_count, mcp_status) = match service_repo.count_running_services("mcp").await {
256 Ok(count) if count > 0 => (count, "healthy"),
257 Ok(_) => (0, "none"),
258 Err(_) => (0, "error"),
259 };
260
261 let web_dir = AppPaths::get()
262 .map(|p| p.web().dist().to_path_buf())
263 .unwrap_or_else(|e| {
264 tracing::debug!(error = %e, "Failed to get web dist path, using default");
265 std::path::PathBuf::from("/var/www/html/dist")
266 });
267 let sitemap_exists = web_dir.join("sitemap.xml").exists();
268 let index_exists = web_dir.join("index.html").exists();
269
270 let db_healthy = db_status == "healthy";
271 let services_ok = agent_status != "error" && mcp_status != "error";
272 let content_ok = sitemap_exists && index_exists;
273
274 let (overall_status, http_status) = match (db_healthy, services_ok && content_ok) {
275 (false, _) => ("unhealthy", StatusCode::SERVICE_UNAVAILABLE),
276 (true, false) => ("degraded", StatusCode::OK),
277 (true, true) => ("healthy", StatusCode::OK),
278 };
279
280 let check_duration_ms = start.elapsed().as_millis();
281 let memory = get_process_memory();
282
283 let data = json!({
284 "status": overall_status,
285 "timestamp": chrono::Utc::now().to_rfc3339(),
286 "version": env!("CARGO_PKG_VERSION"),
287 "checks": {
288 "database": {
289 "status": db_status,
290 "latency_ms": db_latency_ms
291 },
292 "agents": {
293 "status": agent_status,
294 "count": agent_count
295 },
296 "mcp": {
297 "status": mcp_status,
298 "count": mcp_count
299 },
300 "static_content": {
301 "status": if content_ok { "healthy" } else { "degraded" },
302 "index_html": index_exists,
303 "sitemap_xml": sitemap_exists
304 }
305 },
306 "memory": memory,
307 "response_time_ms": check_duration_ms
308 });
309
310 (http_status, Json(data))
311}
312
313pub async fn handle_core_discovery(
314 axum::extract::State(ctx): axum::extract::State<AppContext>,
315) -> impl axum::response::IntoResponse {
316 let base = &ctx.config().api_external_url;
317 let data = json!({
318 "name": "Core Services",
319 "description": "Core conversation, task, and artifact management APIs",
320 "endpoints": {
321 "contexts": {
322 "href": format!("{}{}", base, ApiPaths::CORE_CONTEXTS),
323 "description": "Conversation context management",
324 "methods": ["GET", "POST", "DELETE"]
325 },
326 "tasks": {
327 "href": format!("{}{}", base, ApiPaths::CORE_TASKS),
328 "description": "Task management for agent operations",
329 "methods": ["GET", "POST", "PUT", "DELETE"]
330 },
331 "artifacts": {
332 "href": format!("{}{}", base, ApiPaths::CORE_ARTIFACTS),
333 "description": "Artifact storage and retrieval",
334 "methods": ["GET", "POST", "DELETE"]
335 },
336 "oauth": {
337 "href": format!("{}{}", base, ApiPaths::OAUTH_BASE),
338 "description": "OAuth2/OIDC authentication endpoints"
339 }
340 }
341 });
342 Json(SingleResponse::new(data))
343}
344
345pub async fn handle_agents_discovery(
346 axum::extract::State(ctx): axum::extract::State<AppContext>,
347) -> impl axum::response::IntoResponse {
348 let base = &ctx.config().api_external_url;
349 let data = json!({
350 "name": "Agent Services",
351 "description": "A2A protocol agent registry and proxy",
352 "endpoints": {
353 "registry": {
354 "href": format!("{}{}", base, ApiPaths::AGENTS_REGISTRY),
355 "description": "List and discover available agents",
356 "methods": ["GET"]
357 },
358 "proxy": {
359 "href": format!("{}{}/<agent_id>/", base, ApiPaths::AGENTS_BASE),
360 "description": "Proxy requests to specific agents",
361 "methods": ["GET", "POST"]
362 }
363 }
364 });
365 Json(SingleResponse::new(data))
366}
367
368pub async fn handle_mcp_discovery(
369 axum::extract::State(ctx): axum::extract::State<AppContext>,
370) -> impl axum::response::IntoResponse {
371 let base = &ctx.config().api_external_url;
372 let data = json!({
373 "name": "MCP Services",
374 "description": "Model Context Protocol server registry and proxy",
375 "endpoints": {
376 "registry": {
377 "href": format!("{}{}", base, ApiPaths::MCP_REGISTRY),
378 "description": "List and discover available MCP servers",
379 "methods": ["GET"]
380 },
381 "proxy": {
382 "href": format!("{}{}/<server_name>/mcp", base, ApiPaths::MCP_BASE),
383 "description": "Proxy requests to specific MCP servers",
384 "methods": ["GET", "POST"]
385 }
386 }
387 });
388 Json(SingleResponse::new(data))
389}
390
391pub fn discovery_router(ctx: &AppContext) -> Router {
392 Router::new()
393 .route(ApiPaths::DISCOVERY, get(handle_root_discovery))
394 .route(ApiPaths::HEALTH, get(handle_health))
395 .route("/health", get(handle_health))
396 .route(ApiPaths::CORE_BASE, get(handle_core_discovery))
397 .route(ApiPaths::AGENTS_BASE, get(handle_agents_discovery))
398 .route(ApiPaths::MCP_BASE, get(handle_mcp_discovery))
399 .with_state(ctx.clone())
400}