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

systemprompt_api/routes/agent/
tasks.rs
