a2a_server/
executor.rs

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

a2a_server/
executor.rs
