systemprompt-api 0.15.0

Axum-based HTTP server and API gateway for systemprompt.io AI governance infrastructure. Exposes governed agents, MCP, A2A, and admin endpoints with rate limiting and RBAC.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

use anyhow::Result;
use std::sync::Arc;
use systemprompt_runtime::AppContext;
use systemprompt_traits::{Phase, StartupEvent, StartupEventExt, StartupEventSender};

use super::lifecycle::{
    initialize_scheduler, reconcile_agents, reconcile_system_services, start_event_bridge,
};

pub async fn run_server(ctx: AppContext, events: Option<StartupEventSender>) -> Result<()> {
    let start_time = std::time::Instant::now();

    let mcp_orchestrator = create_mcp_orchestrator(&ctx)?;

    start_event_bridge(&ctx);
    reconcile_system_services(&ctx, &mcp_orchestrator, events.as_ref()).await?;

    if let Some(ref tx) = events {
        tx.phase_started(Phase::Agents);
    }
    match reconcile_agents(&ctx, events.as_ref()).await {
        Ok(started_count) => {
            if let Some(ref tx) = events {
                if tx
                    .unbounded_send(StartupEvent::AgentReconciliationComplete {
                        running: started_count,
                        total: started_count,
                    })
                    .is_err()
                {
                    tracing::debug!("Startup event receiver dropped");
                }
                tx.phase_completed(Phase::Agents);
            }
        },
        Err(e) => {
            if let Some(ref tx) = events {
                tx.phase_failed(Phase::Agents, e.to_string());
                if tx
                    .unbounded_send(StartupEvent::Error {
                        message: format!("Agent reconciliation failed: {e}"),
                        fatal: true,
                    })
                    .is_err()
                {
                    tracing::debug!("Startup event receiver dropped");
                }
            }
            return Err(e);
        },
    }

    if let Some(ref tx) = events {
        tx.phase_started(Phase::Scheduler);
    }
    let scheduler_handle = match initialize_scheduler(&ctx, events.as_ref()).await {
        Ok(handle) => {
            if let Some(ref tx) = events {
                tx.phase_completed(Phase::Scheduler);
            }
            handle
        },
        Err(e) => {
            if let Some(ref tx) = events {
                tx.phase_failed(Phase::Scheduler, e.to_string());
            }
            None
        },
    };

    if let Some(ref tx) = events {
        tx.phase_started(Phase::ApiServer);
    }
    let api_server = crate::services::server::setup_api_server(&ctx, events.clone())?;
    let addr = ctx.server_address();

    if let Some(ref tx) = events {
        tx.phase_completed(Phase::ApiServer);
    }

    if let Some(ref tx) = events {
        tx.startup_complete(start_time.elapsed(), format!("http://{}", addr), vec![]);
    }

    systemprompt_logging::set_startup_mode(false);

    let serve_result = api_server
        .serve(&addr, super::shutdown::shutdown_signal())
        .await;

    super::shutdown::drain(&ctx, scheduler_handle).await;

    serve_result
}

fn create_mcp_orchestrator(
    ctx: &AppContext,
) -> Result<Arc<systemprompt_mcp::services::McpOrchestrator>> {
    use systemprompt_mcp::services::McpOrchestrator;
    let manager = McpOrchestrator::new(
        Arc::clone(ctx.db_pool()),
        Arc::clone(ctx.app_paths_arc()),
        ctx.mcp_registry().clone(),
    )?;
    Ok(Arc::new(manager))
}