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

systemprompt_api/routes/agent/
tasks.rs
