systemprompt-api 0.1.18

HTTP API server and gateway for systemprompt.io OS
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

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};

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)?;
    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);
    }
    match initialize_scheduler(&ctx, events.as_ref()).await {
        Ok(()) => {
            if let Some(ref tx) = events {
                tx.phase_completed(Phase::Scheduler);
            }
        },
        Err(e) => {
            if let Some(ref tx) = events {
                tx.phase_failed(Phase::Scheduler, e.to_string());
            }
        },
    }

    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);

    api_server.serve(&addr).await
}

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