Merge branch 'main' of github.com:crewAIInc/crewAI into lorenze/experimental-environment-tools

lib/crewai/src/crewai/a2a/config.py

@@ -5,7 +5,7 @@ This module is separate from experimental.a2a to avoid circular imports.
 
 from __future__ import annotations
 
-from typing import Annotated, Any, ClassVar
+from typing import Annotated, Any, ClassVar, Literal
 
 from pydantic import (
     BaseModel,
@@ -53,6 +53,7 @@ class A2AConfig(BaseModel):
         fail_fast: If True, raise error when agent unreachable; if False, skip and continue.
         trust_remote_completion_status: If True, return A2A agent's result directly when completed.
         updates: Update mechanism config.
+        transport_protocol: A2A transport protocol (grpc, jsonrpc, http+json).
     """
 
     model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid")
@@ -82,3 +83,7 @@ class A2AConfig(BaseModel):
         default_factory=_get_default_update_config,
         description="Update mechanism config",
     )
+    transport_protocol: Literal["JSONRPC", "GRPC", "HTTP+JSON"] = Field(
+        default="JSONRPC",
+        description="Specified mode of A2A transport protocol",
+    )

lib/crewai/src/crewai/a2a/utils.py

@@ -7,7 +7,7 @@ from collections.abc import AsyncIterator, MutableMapping
 from contextlib import asynccontextmanager
 from functools import lru_cache
 import time
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal
 import uuid
 
 from a2a.client import A2AClientHTTPError, Client, ClientConfig, ClientFactory
@@ -18,7 +18,6 @@ from a2a.types import (
     PushNotificationConfig as A2APushNotificationConfig,
     Role,
     TextPart,
-    TransportProtocol,
 )
 from aiocache import cached  # type: ignore[import-untyped]
 from aiocache.serializers import PickleSerializer  # type: ignore[import-untyped]
@@ -259,6 +258,7 @@ async def _afetch_agent_card_impl(
 
 def execute_a2a_delegation(
     endpoint: str,
+    transport_protocol: Literal["JSONRPC", "GRPC", "HTTP+JSON"],
     auth: AuthScheme | None,
     timeout: int,
     task_description: str,
@@ -282,6 +282,23 @@ def execute_a2a_delegation(
     use aexecute_a2a_delegation directly.
 
     Args:
+        endpoint: A2A agent endpoint URL (AgentCard URL)
+        transport_protocol: Optional A2A transport protocol (grpc, jsonrpc, http+json)
+        auth: Optional AuthScheme for authentication (Bearer, OAuth2, API Key, HTTP Basic/Digest)
+        timeout: Request timeout in seconds
+        task_description: The task to delegate
+        context: Optional context information
+        context_id: Context ID for correlating messages/tasks
+        task_id: Specific task identifier
+        reference_task_ids: List of related task IDs
+        metadata: Additional metadata (external_id, request_id, etc.)
+        extensions: Protocol extensions for custom fields
+        conversation_history: Previous Message objects from conversation
+        agent_id: Agent identifier for logging
+        agent_role: Role of the CrewAI agent delegating the task
+        agent_branch: Optional agent tree branch for logging
+        response_model: Optional Pydantic model for structured outputs
+        turn_number: Optional turn number for multi-turn conversations
         endpoint: A2A agent endpoint URL.
         auth: Optional AuthScheme for authentication.
         timeout: Request timeout in seconds.
@@ -323,6 +340,7 @@ def execute_a2a_delegation(
                 agent_role=agent_role,
                 agent_branch=agent_branch,
                 response_model=response_model,
+                transport_protocol=transport_protocol,
                 turn_number=turn_number,
                 updates=updates,
             )
@@ -333,6 +351,7 @@ def execute_a2a_delegation(
 
 async def aexecute_a2a_delegation(
     endpoint: str,
+    transport_protocol: Literal["JSONRPC", "GRPC", "HTTP+JSON"],
     auth: AuthScheme | None,
     timeout: int,
     task_description: str,
@@ -356,6 +375,23 @@ async def aexecute_a2a_delegation(
     in an async context (e.g., with Crew.akickoff() or agent.aexecute_task()).
 
     Args:
+        endpoint: A2A agent endpoint URL
+        transport_protocol: Optional A2A transport protocol (grpc, jsonrpc, http+json)
+        auth: Optional AuthScheme for authentication
+        timeout: Request timeout in seconds
+        task_description: Task to delegate
+        context: Optional context
+        context_id: Context ID for correlation
+        task_id: Specific task identifier
+        reference_task_ids: Related task IDs
+        metadata: Additional metadata
+        extensions: Protocol extensions
+        conversation_history: Previous Message objects
+        turn_number: Current turn number
+        agent_branch: Agent tree branch for logging
+        agent_id: Agent identifier for logging
+        agent_role: Agent role for logging
+        response_model: Optional Pydantic model for structured outputs
         endpoint: A2A agent endpoint URL.
         auth: Optional AuthScheme for authentication.
         timeout: Request timeout in seconds.
@@ -414,6 +450,7 @@ async def aexecute_a2a_delegation(
         agent_role=agent_role,
         response_model=response_model,
         updates=updates,
+        transport_protocol=transport_protocol,
     )
 
     crewai_event_bus.emit(
@@ -431,6 +468,7 @@ async def aexecute_a2a_delegation(
 
 async def _aexecute_a2a_delegation_impl(
     endpoint: str,
+    transport_protocol: Literal["JSONRPC", "GRPC", "HTTP+JSON"],
     auth: AuthScheme | None,
     timeout: int,
     task_description: str,
@@ -524,7 +562,6 @@ async def _aexecute_a2a_delegation_impl(
         extensions=extensions,
     )
 
-    transport_protocol = TransportProtocol("JSONRPC")
     new_messages: list[Message] = [*conversation_history, message]
     crewai_event_bus.emit(
         None,
@@ -596,7 +633,7 @@ async def _aexecute_a2a_delegation_impl(
 @asynccontextmanager
 async def _create_a2a_client(
     agent_card: AgentCard,
-    transport_protocol: TransportProtocol,
+    transport_protocol: Literal["JSONRPC", "GRPC", "HTTP+JSON"],
     timeout: int,
     headers: MutableMapping[str, str],
     streaming: bool,
@@ -640,7 +677,7 @@ async def _create_a2a_client(
 
         config = ClientConfig(
             httpx_client=httpx_client,
-            supported_transports=[str(transport_protocol.value)],
+            supported_transports=[transport_protocol],
             streaming=streaming and not use_polling,
             polling=use_polling,
             accepted_output_modes=["application/json"],

lib/crewai/src/crewai/a2a/wrapper.py

@@ -771,6 +771,7 @@ def _delegate_to_a2a(
                 response_model=agent_config.response_model,
                 turn_number=turn_num + 1,
                 updates=agent_config.updates,
+                transport_protocol=agent_config.transport_protocol,
             )
 
             conversation_history = a2a_result.get("history", [])
@@ -1085,6 +1086,7 @@ async def _adelegate_to_a2a(
                 agent_branch=agent_branch,
                 response_model=agent_config.response_model,
                 turn_number=turn_num + 1,
+                transport_protocol=agent_config.transport_protocol,
                 updates=agent_config.updates,
             )
 

lib/crewai/src/crewai/events/event_listener.py

@@ -209,10 +209,9 @@ class EventListener(BaseEventListener):
         @crewai_event_bus.on(TaskCompletedEvent)
         def on_task_completed(source: Any, event: TaskCompletedEvent) -> None:
             # Handle telemetry
-            span = self.execution_spans.get(source)
+            span = self.execution_spans.pop(source, None)
             if span:
                 self._telemetry.task_ended(span, source, source.agent.crew)
-            self.execution_spans[source] = None
 
             # Pass task name if it exists
             task_name = get_task_name(source)
@@ -222,11 +221,10 @@ class EventListener(BaseEventListener):
 
         @crewai_event_bus.on(TaskFailedEvent)
         def on_task_failed(source: Any, event: TaskFailedEvent) -> None:
-            span = self.execution_spans.get(source)
+            span = self.execution_spans.pop(source, None)
             if span:
                 if source.agent and source.agent.crew:
                     self._telemetry.task_ended(span, source, source.agent.crew)
-                self.execution_spans[source] = None
 
             # Pass task name if it exists
             task_name = get_task_name(source)