a2a_server/
executor.rs

1// Copyright AGNTCY Contributors (https://github.com/agntcy)
2// SPDX-License-Identifier: Apache-2.0
3use a2a::*;
4use futures::stream::BoxStream;
5use serde_json::Value;
6use std::collections::HashMap;
7
8use crate::middleware::{ServiceParams, User};
9
10/// Context passed to the [`AgentExecutor`] for each execution.
11pub struct ExecutorContext {
12    /// The message that triggered execution (None for cancellation without message).
13    pub message: Option<Message>,
14    /// The task ID being executed.
15    pub task_id: TaskId,
16    /// The existing task, if this is a continuation.
17    pub stored_task: Option<Task>,
18    /// Context ID grouping related interactions.
19    pub context_id: String,
20    /// Request metadata.
21    pub metadata: Option<HashMap<String, Value>>,
22    /// Authenticated user info.
23    pub user: Option<User>,
24    /// Service parameters (headers/metadata) from the request.
25    pub service_params: ServiceParams,
26    /// Optional tenant ID.
27    pub tenant: Option<String>,
28}
29
30impl ExecutorContext {
31    pub fn task_info(&self) -> (TaskId, String) {
32        (self.task_id.clone(), self.context_id.clone())
33    }
34}
35
36/// The user's business logic entry point.
37///
38/// Implement this trait to handle incoming messages and cancellation requests.
39/// The executor returns a stream of events (status updates, artifact updates,
40/// messages, or task objects).
41#[async_trait::async_trait]
42pub trait AgentExecutor: Send + Sync + 'static {
43    /// Execute a message. Returns a stream of events.
44    fn execute(&self, ctx: ExecutorContext)
45    -> BoxStream<'static, Result<StreamResponse, A2AError>>;
46
47    /// Cancel an in-progress task. Returns a stream of events.
48    fn cancel(&self, ctx: ExecutorContext) -> BoxStream<'static, Result<StreamResponse, A2AError>>;
49}
50
51#[cfg(test)]
52mod tests {
53    use super::*;
54
55    #[test]
56    fn test_executor_context_task_info() {
57        let ctx = ExecutorContext {
58            message: None,
59            task_id: "t1".into(),
60            stored_task: None,
61            context_id: "c1".into(),
62            metadata: None,
63            user: None,
64            service_params: HashMap::new(),
65            tenant: None,
66        };
67        let (tid, cid) = ctx.task_info();
68        assert_eq!(tid, "t1");
69        assert_eq!(cid, "c1");
70    }
71}
a2a_server/executor.rs

a2a_server/
executor.rs
