Trait TaskStore 

pub trait TaskStore:
    Send
    + Sync
    + 'static {
    // Required methods
    fn create_task<'life0, 'life1, 'async_trait>(
        &'life0 self,
        tool_name: &'life1 str,
        arguments: Value,
        ttl: Option<u64>,
        owner: TaskOwner,
    ) -> Pin<Box<dyn Future<Output = Result<(String, CancellationToken)>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn task_owner<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<TaskOwner>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn get_task<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<TaskObject>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn get_task_result<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<TaskSnapshot>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn wait_for_completion<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<TaskSnapshot>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn list_tasks<'life0, 'async_trait>(
        &'life0 self,
        status_filter: Option<TaskStatus>,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<TaskObject>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn require_input<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
        requests: InputRequests,
        message: Option<&'life2 str>,
    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;
    fn outstanding_input_requests<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<InputRequests>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn apply_input_responses<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
        responses: InputResponses,
    ) -> Pin<Box<dyn Future<Output = Result<Option<AppliedInputResponses>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn set_ttl<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
        ttl_ms: u64,
    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn complete_task<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
        result: CallToolResult,
    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn fail_task<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
        error: JsonRpcError,
    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn cancel_task<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
        reason: Option<&'life2 str>,
    ) -> Pin<Box<dyn Future<Output = Result<Option<TaskObject>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;

    // Provided methods
    fn set_task_meta<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
        meta: Value,
    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    fn discard_task<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    fn task_presence<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<TaskPresence>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    fn input_responses<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<InputResponses>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    fn set_status<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
        status: TaskStatus,
        message: Option<&'life2 str>,
    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait { ... }
    fn resume_context<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<TaskResumeContext>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
}

Storage backend for async task state.

Implementations persist task lifecycle state keyed by task ID. The default implementation is MemoryTaskStore; external stores (Redis, Postgres, etc.) typically live in separate crates.

Semantics

• Terminal states (TaskStatus::is_terminal) are immutable: once a task is completed, failed, or cancelled, further transitions must be rejected (Ok(false) from the transition methods).
• An expired task is indistinguishable from an unknown one. Reads return None once ttlMs has elapsed since creation, whether or not the entry has actually been reclaimed, so callers cannot probe for the existence of a task whose retention window has closed.
• cancel_task and TTL expiry must signal the task’s CancellationToken even if the task is already terminal. The token is a persistent signal and may be awaited, so it must be the same cancellation domain returned from creation and carried through TaskResumeContext.
• wait_for_completion blocks until the task reaches a terminal state or expires. Expiry wakes an existing waiter, which then returns None; how an implementation waits (notification, polling, pub/sub) is an implementation detail and must not leak into the trait.

Implementing this trait

Three methods carry defaults so that stores written before the features existed keep compiling: set_task_meta, discard_task, and resume_context. Each default reports “not supported” rather than quietly succeeding, and the router turns that into a visible failure. A store that supports input requests must therefore override resume_context, or its first tasks/update fails the task.

That also makes wrapping another store a trap worth naming. A decorator that implements only the required methods inherits the defaults, which silently disables resumption for the store it wraps even though the wrapped store supports it. Forward every method, including the defaulted ones:

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

use async_trait::async_trait;
use tower_mcp::CallToolResult;
use tower_mcp::async_task::{
    AppliedInputResponses, CancellationToken, MemoryTaskStore, Result, TaskOwner,
    TaskResumeContext, TaskSnapshot, TaskStore,
    TaskPresence,
};
use tower_mcp::error::JsonRpcError;
use tower_mcp::protocol::{InputRequests, InputResponses, TaskObject, TaskStatus};

/// Counts the completions it actually applied, and delegates the rest.
struct CountingStore {
    inner: MemoryTaskStore,
    completed: Arc<AtomicUsize>,
}

#[async_trait]
impl TaskStore for CountingStore {
    async fn complete_task(&self, id: &str, result: CallToolResult) -> Result<bool> {
        let applied = self.inner.complete_task(id, result).await?;
        // `false` means the task was already terminal, expired, or gone,
        // so counting it would inflate the number of finished tasks.
        if applied {
            self.completed.fetch_add(1, Ordering::Relaxed);
        }
        Ok(applied)
    }

    // Required methods, forwarded unchanged.
    async fn create_task(
        &self,
        tool_name: &str,
        arguments: serde_json::Value,
        ttl: Option<u64>,
        owner: TaskOwner,
    ) -> Result<(String, CancellationToken)> {
        self.inner.create_task(tool_name, arguments, ttl, owner).await
    }
    async fn task_owner(&self, id: &str) -> Result<Option<TaskOwner>> {
        self.inner.task_owner(id).await
    }
    // Forward this too when the wrapped store retains tombstones, or the
    // default turns its `Expired` into `Missing` and the decorator
    // silently removes the distinction (#1249).
    async fn task_presence(&self, id: &str) -> Result<TaskPresence> {
        self.inner.task_presence(id).await
    }
    async fn get_task(&self, id: &str) -> Result<Option<TaskObject>> {
        self.inner.get_task(id).await
    }
    async fn get_task_result(&self, id: &str) -> Result<Option<TaskSnapshot>> {
        self.inner.get_task_result(id).await
    }
    async fn wait_for_completion(&self, id: &str) -> Result<Option<TaskSnapshot>> {
        self.inner.wait_for_completion(id).await
    }
    async fn list_tasks(&self, status: Option<TaskStatus>) -> Result<Vec<TaskObject>> {
        self.inner.list_tasks(status).await
    }
    async fn require_input(
        &self,
        id: &str,
        requests: InputRequests,
        message: Option<&str>,
    ) -> Result<bool> {
        self.inner.require_input(id, requests, message).await
    }
    async fn outstanding_input_requests(&self, id: &str) -> Result<Option<InputRequests>> {
        self.inner.outstanding_input_requests(id).await
    }
    async fn apply_input_responses(
        &self,
        id: &str,
        responses: InputResponses,
    ) -> Result<Option<AppliedInputResponses>> {
        self.inner.apply_input_responses(id, responses).await
    }
    async fn set_ttl(&self, id: &str, ttl_ms: u64) -> Result<bool> {
        self.inner.set_ttl(id, ttl_ms).await
    }
    async fn fail_task(&self, id: &str, error: JsonRpcError) -> Result<bool> {
        self.inner.fail_task(id, error).await
    }
    async fn cancel_task(&self, id: &str, reason: Option<&str>) -> Result<Option<TaskObject>> {
        self.inner.cancel_task(id, reason).await
    }

    // Defaulted methods. Omitting these would leave the wrapper reporting
    // "not supported" for a store that supports them.
    async fn resume_context(&self, id: &str) -> Result<Option<TaskResumeContext>> {
        self.inner.resume_context(id).await
    }
    async fn set_task_meta(&self, id: &str, meta: serde_json::Value) -> Result<bool> {
        self.inner.set_task_meta(id, meta).await
    }
    async fn discard_task(&self, id: &str) -> Result<bool> {
        self.inner.discard_task(id).await
    }
}

let completed = Arc::new(AtomicUsize::new(0));
let store: Arc<dyn TaskStore> = Arc::new(CountingStore {
    inner: MemoryTaskStore::new(),
    completed: completed.clone(),
});
// Ready to hand to `McpRouter::task_store`.

let (id, _cancel) = store
    .create_task("build_report", serde_json::json!({}), None, None)
    .await
    .unwrap();
assert!(store.complete_task(&id, CallToolResult::text("done")).await.unwrap());
assert!(!store.complete_task(&id, CallToolResult::text("again")).await.unwrap());
assert_eq!(completed.load(Ordering::Relaxed), 1);

Required Methods

fn create_task<'life0, 'life1, 'async_trait>( &'life0 self, tool_name: &'life1 str, arguments: Value, ttl: Option<u64>, owner: TaskOwner, ) -> Pin<Box<dyn Future<Output = Result<(String, CancellationToken)>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Create and store a new task owned by owner.

Returns the task ID and a cancellation token for the spawned work. The token is awaitable and must be raised both by explicit cancellation and when the task’s TTL elapses. An external store backed by a remote database is responsible for bridging its durable expiry signal to this process-local token. owner is the authenticated principal responsible for the task, or None when the request carried no authenticated context.

fn task_owner<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskOwner>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Read a task’s owner.

The outer Option distinguishes a known task from an unknown or expired one; the inner TaskOwner distinguishes an owned task from one created without an authenticated principal.

fn get_task<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskObject>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Get task object by ID. Returns None if unknown.

fn get_task_result<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskSnapshot>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Get a task’s full snapshot (task object, result, error) by ID.

fn wait_for_completion<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskSnapshot>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Wait for a task to reach a terminal state, then return its snapshot.

If the task is already terminal, returns immediately. Otherwise blocks until the task completes, fails, is cancelled, or expires. Returns None if the task is unknown or expires while waiting.

fn list_tasks<'life0, 'async_trait>( &'life0 self, status_filter: Option<TaskStatus>, ) -> Pin<Box<dyn Future<Output = Result<Vec<TaskObject>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

List all tasks, optionally filtered by status.

fn require_input<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, task_id: &'life1 str, requests: InputRequests, message: Option<&'life2 str>, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Mark a task as requiring input, recording the requests to be answered.

requests replaces the outstanding set. A key that was outstanding and does not appear in the new snapshot becomes superseded.

Returns Ok(false) if the task is unknown, expired, or already terminal.

Key uniqueness

SEP-2663 requires every request key to be unique over a single task’s lifetime. A key is spent once its request has been answered or superseded, and must never name a second request; that guarantee is what lets a client deduplicate across polls and lets a server ignore responses for already-satisfied requests (#1246).

Reissuing a spent key is a TaskStoreError::InvalidTransition, not a backend failure: it is deterministic, and retrying cannot help.

Carrying an unanswered request forward is not reuse. Because requests replaces the whole snapshot, a still-outstanding key has to be reissued to stay outstanding, and doing so names the same question rather than a second one. Repointing a live key at a different request is reuse and must be rejected.

The uniqueness scope is the task. Keys from a preceding MRTR phase are a separate namespace and do not constrain a task’s keys.

Implementing this

An external store must enforce the same rule, and the check and the snapshot replacement must be atomic: two concurrent parks that each see a key as unspent would otherwise both admit it. Stores written before this rule existed keep compiling and keep their old permissive behaviour, so this is a behavioural migration rather than a compile-visible one.

Example

use tower_mcp::async_task::{MemoryTaskStore, TaskStore, TaskStoreError};
use tower_mcp::protocol::{InputRequest, InputRequests, ListRootsParams, TaskStatus};

fn ask(keys: &[&str]) -> InputRequests {
    keys.iter()
        .map(|key| {
            (
                key.to_string(),
                InputRequest::ListRoots(ListRootsParams { meta: None }),
            )
        })
        .collect()
}

let store = MemoryTaskStore::new();
let (id, _cancel) = store
    .create_task("deploy", serde_json::json!({}), None, None)
    .await
    .unwrap();

assert!(
    store
        .require_input(&id, ask(&["approval"]), Some("needs a decision"))
        .await
        .unwrap()
);
let task = store.get_task(&id).await.unwrap().unwrap();
assert_eq!(task.status, TaskStatus::InputRequired);
assert_eq!(task.status_message.as_deref(), Some("needs a decision"));

// Asking a second thing means reissuing the first: the snapshot is
// replaced wholesale, so a key left out becomes superseded.
assert!(
    store
        .require_input(&id, ask(&["approval", "region"]), None)
        .await
        .unwrap()
);

// A spent key cannot come back. This one was superseded rather than
// answered; both count as spent.
store
    .require_input(&id, ask(&["region"]), None)
    .await
    .unwrap();
let error = store
    .require_input(&id, ask(&["approval"]), None)
    .await
    .expect_err("a spent key must not name a second question");
assert!(matches!(error, TaskStoreError::InvalidTransition(_)));

fn outstanding_input_requests<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<InputRequests>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Read the requests a task is currently waiting on.

Returns an empty map when the task is not input_required, and None when the task is unknown or expired.

fn apply_input_responses<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, responses: InputResponses, ) -> Pin<Box<dyn Future<Output = Result<Option<AppliedInputResponses>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Apply tasks/update.inputResponses to a task.

Consumes the keys that match an outstanding request and ignores the rest. When the last outstanding request is answered the task returns to TaskStatus::Working.

Returns None if the task is unknown, expired, or already terminal.

Ignoring is deliberate rather than lenient parsing. A key that was never issued, one already answered, and one superseded by a later request are indistinguishable to a client that is retrying, and rejecting the whole update would fail a task over a duplicate delivery. They are reported in AppliedInputResponses::ignored so a server can still notice.

Example

use tower_mcp::async_task::{MemoryTaskStore, TaskStore};
use tower_mcp::protocol::{
    ElicitAction, ElicitResult, InputRequest, InputRequests, InputResponse,
    ListRootsParams, TaskStatus,
};

fn ask(keys: &[&str]) -> InputRequests {
    keys.iter()
        .map(|key| {
            (
                key.to_string(),
                InputRequest::ListRoots(ListRootsParams { meta: None }),
            )
        })
        .collect()
}

fn accept(key: &str) -> (String, InputResponse) {
    (
        key.to_string(),
        InputResponse::Elicit(ElicitResult {
            action: ElicitAction::Accept,
            content: None,
            meta: None,
        }),
    )
}

let store = MemoryTaskStore::new();
let (id, _cancel) = store
    .create_task("deploy", serde_json::json!({}), None, None)
    .await
    .unwrap();
store
    .require_input(&id, ask(&["approval", "region"]), None)
    .await
    .unwrap();

// A partial answer is valid, and leaves the task parked.
let applied = store
    .apply_input_responses(&id, [accept("approval")].into_iter().collect())
    .await
    .unwrap()
    .unwrap();
assert_eq!(applied.accepted, ["approval".to_string()].into());
assert_eq!(applied.still_outstanding, ["region".to_string()].into());
assert!(!applied.is_complete());
assert_eq!(
    store.get_task(&id).await.unwrap().unwrap().status,
    TaskStatus::InputRequired
);

// A replayed answer, plus a key nobody asked for: both ignored, so the
// task is neither failed nor resumed early.
let applied = store
    .apply_input_responses(
        &id,
        [accept("approval"), accept("never-issued")].into_iter().collect(),
    )
    .await
    .unwrap()
    .unwrap();
assert!(applied.accepted.is_empty());
assert_eq!(
    applied.ignored,
    ["approval".to_string(), "never-issued".to_string()].into()
);

// Answering the last outstanding request is what resumes the task.
let applied = store
    .apply_input_responses(&id, [accept("region")].into_iter().collect())
    .await
    .unwrap()
    .unwrap();
assert!(applied.is_complete());
assert_eq!(
    store.get_task(&id).await.unwrap().unwrap().status,
    TaskStatus::Working
);

fn set_ttl<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ttl_ms: u64, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Update a task’s time-to-live, measured from creation.

SEP-2663 allows ttlMs to change over a task’s lifetime. Returns Ok(false) if the task is unknown or already expired.

fn complete_task<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, result: CallToolResult, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Mark a task as completed with a result.

A result carrying isError: true still completes the task: the tool ran and produced a domain error, which SEP-2663 distinguishes from an execution failure.

Returns Ok(false) if the task is unknown, expired, or already terminal.

MemoryTaskStore returns TaskStoreError::RetentionLimitExceeded for an oversized result only after atomically replacing the live record with a fixed, content-free failed snapshot and waking completion waiters. The rejected result is not retained. External stores may choose a different recovery policy, so generic callers should still handle the error rather than assuming every implementation terminalized.

Example

A tool that ran and reported a problem is a completed task, and the error result is what tasks/get hands back:

use tower_mcp::CallToolResult;
use tower_mcp::async_task::{MemoryTaskStore, TaskStore};
use tower_mcp::protocol::TaskStatus;

let store = MemoryTaskStore::new();
let (id, _cancel) = store
    .create_task("deploy", serde_json::json!({}), None, None)
    .await
    .unwrap();

let mut result = CallToolResult::text("region eu-west-3 is not enabled");
result.is_error = true;
assert!(store.complete_task(&id, result).await.unwrap());

let (task, result, error) = store.get_task_result(&id).await.unwrap().unwrap();
assert_eq!(task.status, TaskStatus::Completed);
assert!(result.unwrap().is_error);
assert!(error.is_none(), "isError is a result, not a JSON-RPC error");

fn fail_task<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, error: JsonRpcError, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Mark a task as failed with a structured execution error.

Returns Ok(false) if the task is unknown, expired, or already terminal.

MemoryTaskStore returns TaskStoreError::RetentionLimitExceeded for an oversized error only after atomically storing its fixed, content-free failed snapshot and waking completion waiters. The rejected diagnostic is not retained.

Reserved for a call that never produced a result at all. A tool that ran and reported a problem completes instead, carrying an isError result; see complete_task.

Example

The error is stored whole, not flattened to a message, because SEP-2663 requires tasks/get on a failed task to return a JSON-RPC error object:

use tower_mcp::async_task::{MemoryTaskStore, TaskStore};
use tower_mcp::error::JsonRpcError;
use tower_mcp::protocol::TaskStatus;

let store = MemoryTaskStore::new();
let (id, _cancel) = store
    .create_task("deploy", serde_json::json!({}), None, None)
    .await
    .unwrap();

let mut error = JsonRpcError::invalid_params("unknown region");
error.data = Some(serde_json::json!({"field": "region"}));
assert!(store.fail_task(&id, error).await.unwrap());

let (task, result, error) = store.get_task_result(&id).await.unwrap().unwrap();
assert_eq!(task.status, TaskStatus::Failed);
assert!(result.is_none());

let error = error.unwrap();
assert_eq!(error.code, -32602);
assert_eq!(error.data.unwrap()["field"], "region");

fn cancel_task<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, task_id: &'life1 str, reason: Option<&'life2 str>, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskObject>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Cancel a task.

Signals the task’s CancellationToken and, if the task is not already terminal, marks it cancelled. Returns the updated task object, or None if the task is unknown.

The token is raised even for a task that already finished, so work still winding down behind a completed task is told to stop. That is also why the returned object may read completed rather than cancelled: the recorded outcome does not change, only the token.

Example

use tower_mcp::CallToolResult;
use tower_mcp::async_task::{MemoryTaskStore, TaskStore};
use tower_mcp::protocol::TaskStatus;

let store = MemoryTaskStore::new();
let (id, token) = store
    .create_task("deploy", serde_json::json!({}), None, None)
    .await
    .unwrap();

// A handler polls the token between steps; cancellation is
// cooperative and interrupts nothing on its own.
assert!(!token.is_cancelled());

let task = store.cancel_task(&id, Some("user closed the tab")).await.unwrap();
assert_eq!(task.unwrap().status, TaskStatus::Cancelled);
assert!(token.is_cancelled());

// Cancelling again is harmless, and a late result is refused.
assert!(
    !store
        .complete_task(&id, CallToolResult::text("finished anyway"))
        .await
        .unwrap()
);

Provided Methods

fn set_task_meta<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, meta: Value, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Persist protocol _meta for a task.

The default preserves source compatibility for external stores. Stores that want to support task preparation metadata must override it.

fn discard_task<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Remove a task that could not finish initialization.

The default preserves source compatibility for external stores. Stores used with preparation callbacks should override it.

fn task_presence<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<TaskPresence>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Resolve a task to present, expired, or missing, with its owner.

The router uses this both to authorize an operation and to classify an absent result afterwards, so a store that retains expired records can tell its owner “that task expired” instead of “no such task” (#1249).

Defaults to today’s behaviour, treating anything task_owner does not return as TaskPresence::Missing, so existing stores compile and behave unchanged. Override it when the store retains tombstones.

Implementing this

Report Expired only for a record the store still holds; dropping expired records and answering Missing is correct.

The owner must be returned for Expired as well as Present. The router authorizes before disclosing the difference, and it cannot do that without knowing who owns an expired task.

The lookup should be atomic with respect to expiry where the backend allows it: the router resolves a second time after an operation returns nothing, and a store whose active and tombstone lookups can disagree may answer Missing for a task that expired mid-operation.

fn input_responses<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<InputResponses>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Every answer accumulated for a task so far, keyed as issued.

A live handler reads this after being woken, so that what it observes is exactly what was durably recorded (#1246). Defaults to reading through resume_context, so a store that already supports resumption needs no change.

fn set_status<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, task_id: &'life1 str, status: TaskStatus, message: Option<&'life2 str>, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Set a task’s non-terminal status and message.

Terminal states are reached through complete_task, fail_task, and cancel_task; this is for progress reporting while a task is still running. Returns Ok(false) if the task is unknown, expired, or already terminal.

fn resume_context<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskResumeContext>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Everything needed to re-invoke a task’s handler after its input requests were answered.

Returning None means this store cannot resume, and the router fails the task with a message saying so rather than leaving it in working forever. The default returns None so an external store written before resumption existed keeps compiling and fails loudly instead of hanging; implement it to support the flow (#1208).

The returned context must also carry the task’s cancellation/expiry signal. Construct it with TaskResumeContext::new and attach the token with TaskResumeContext::with_cancellation_token. Reusing the signal returned from create_task lets the router stop a replayed handler exactly when the task expires.

Example

The answers accumulate across rounds, because the handler is re-run from the top and has to see everything it was told so far, not only the most recent answer:

use tower_mcp::async_task::{MemoryTaskStore, TaskStore};
use tower_mcp::protocol::{
    ElicitAction, ElicitResult, InputRequest, InputRequests, InputResponse,
    ListRootsParams,
};

let store = MemoryTaskStore::new();
let (id, _cancel) = store
    .create_task("deploy", serde_json::json!({"service": "api"}), None, None)
    .await
    .unwrap();

store.require_input(&id, ask(&["approval"]), None).await.unwrap();
store
    .apply_input_responses(&id, [accept("approval")].into_iter().collect())
    .await
    .unwrap();
store.require_input(&id, ask(&["region"]), None).await.unwrap();
store
    .apply_input_responses(&id, [accept("region")].into_iter().collect())
    .await
    .unwrap();

let resume = store.resume_context(&id).await.unwrap().unwrap();
assert_eq!(resume.tool_name, "deploy");
assert_eq!(resume.arguments, serde_json::json!({"service": "api"}));
assert_eq!(
    resume.input_responses.keys().collect::<Vec<_>>(),
    vec!["approval", "region"],
    "an earlier round's answer is still there on the second resume"
);

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl TaskStore for MemoryTaskStore