systemprompt_api/routes/agent/
tasks.rs

1//! Agent task listing and lookup endpoints.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use axum::extract::{Path, Query, State};
7use axum::http::StatusCode;
8use axum::response::IntoResponse;
9use axum::{Extension, Json};
10use serde::Deserialize;
11use systemprompt_identifiers::{ContextId, TaskId, UserId};
12
13use systemprompt_agent::models::a2a::TaskState;
14use systemprompt_agent::repository::context::ContextRepository;
15use systemprompt_agent::repository::task::TaskRepository;
16use systemprompt_models::RequestContext;
17use systemprompt_runtime::AppContext;
18
19use crate::error::ApiHttpError;
20
21#[derive(Debug, Deserialize)]
22pub struct TaskFilterParams {
23    pub status: Option<String>,
24    pub limit: Option<u32>,
25}
26
27pub async fn list_tasks_by_context(
28    Extension(req_ctx): Extension<RequestContext>,
29    State(app_context): State<AppContext>,
30    Path(context_id): Path<String>,
31) -> Result<impl IntoResponse, ApiHttpError> {
32    tracing::debug!(context_id = %context_id, "Listing tasks");
33
34    let context_id_typed = ContextId::new(&context_id);
35
36    let context_repo = ContextRepository::new(app_context.db_pool())?;
37    context_repo
38        .validate_context_ownership(&context_id_typed, req_ctx.user_id())
39        .await?;
40
41    let task_repo = TaskRepository::new(app_context.db_pool())?;
42    let tasks = task_repo.list_tasks_by_context(&context_id_typed).await?;
43
44    tracing::debug!(context_id = %context_id, count = %tasks.len(), "Tasks listed");
45    Ok((StatusCode::OK, Json(tasks)))
46}
47
48pub async fn get_task(
49    Extension(req_ctx): Extension<RequestContext>,
50    State(app_context): State<AppContext>,
51    Path(task_id): Path<String>,
52) -> Result<impl IntoResponse, ApiHttpError> {
53    tracing::debug!(task_id = %task_id, "Retrieving task");
54
55    let task_repo = TaskRepository::new(app_context.db_pool())?;
56
57    let task_id_typed = TaskId::new(&task_id);
58    task_repo
59        .validate_task_ownership(&task_id_typed, req_ctx.user_id())
60        .await?;
61
62    let task = task_repo
63        .get_task(&task_id_typed)
64        .await?
65        .ok_or_else(|| ApiHttpError::not_found(format!("Task '{task_id}' not found")))?;
66
67    tracing::debug!("Task retrieved successfully");
68    Ok((StatusCode::OK, Json(task)))
69}
70
71pub async fn list_tasks_by_user(
72    Extension(req_ctx): Extension<RequestContext>,
73    State(app_context): State<AppContext>,
74    Query(params): Query<TaskFilterParams>,
75) -> Result<impl IntoResponse, ApiHttpError> {
76    let user_id = req_ctx.auth.actor.user_id.as_str();
77
78    tracing::debug!(user_id = %user_id, "Listing tasks");
79
80    let task_repo = TaskRepository::new(app_context.db_pool())?;
81
82    let task_state = params.status.as_ref().and_then(|s| match s.as_str() {
83        "submitted" => Some(TaskState::Submitted),
84        "working" => Some(TaskState::Working),
85        "input-required" => Some(TaskState::InputRequired),
86        "completed" => Some(TaskState::Completed),
87        "canceled" | "cancelled" => Some(TaskState::Canceled),
88        "failed" => Some(TaskState::Failed),
89        "rejected" => Some(TaskState::Rejected),
90        "auth-required" => Some(TaskState::AuthRequired),
91        _ => None,
92    });
93
94    let user_id_typed = UserId::new(user_id);
95    let mut tasks = task_repo
96        .get_tasks_by_user_id(&user_id_typed, params.limit.map(|l| l as i32), None)
97        .await?;
98
99    if let Some(state) = task_state {
100        tasks.retain(|t| t.status.state == state);
101    }
102
103    tracing::debug!(user_id = %user_id, count = %tasks.len(), "Tasks listed");
104    Ok((StatusCode::OK, Json(tasks)))
105}
106
107pub async fn get_messages_by_task(
108    Extension(req_ctx): Extension<RequestContext>,
109    State(app_context): State<AppContext>,
110    Path(task_id): Path<String>,
111) -> Result<impl IntoResponse, ApiHttpError> {
112    tracing::debug!(task_id = %task_id, "Retrieving messages");
113
114    let task_repo = TaskRepository::new(app_context.db_pool())?;
115
116    let task_id_typed = TaskId::new(&task_id);
117    task_repo
118        .validate_task_ownership(&task_id_typed, req_ctx.user_id())
119        .await?;
120
121    let messages = task_repo.get_messages_by_task(&task_id_typed).await?;
122
123    tracing::debug!(task_id = %task_id, count = %messages.len(), "Messages retrieved");
124    Ok((StatusCode::OK, Json(messages)))
125}
126
127pub async fn delete_task(
128    Extension(req_ctx): Extension<RequestContext>,
129    State(app_context): State<AppContext>,
130    Path(task_id): Path<String>,
131) -> Result<impl IntoResponse, ApiHttpError> {
132    tracing::debug!(task_id = %task_id, "Deleting task");
133
134    let task_repo = TaskRepository::new(app_context.db_pool())?;
135
136    let task_id_typed = TaskId::new(&task_id);
137    task_repo
138        .validate_task_ownership(&task_id_typed, req_ctx.user_id())
139        .await?;
140
141    task_repo.delete_task(&task_id_typed).await?;
142
143    tracing::debug!(task_id = %task_id, "Task deleted");
144    Ok(StatusCode::NO_CONTENT)
145}
systemprompt_api/routes/agent/tasks.rs

systemprompt_api/routes/agent/
tasks.rs
