A2A 的类型定义把"协议是标准"这件事钉死（a2a/types.py:37）：

# a2a/types.py:37
TransportType = Literal["JSONRPC", "GRPC", "HTTP+JSON"]      # 三种传输方式
ProtocolVersion = Literal["0.2.0","0.2.1",...,"0.3.0","0.4.0"]  # 明确的协议版本枚举

# :60 远端 Agent 回复的协议形状
@runtime_checkable
class AgentResponseProtocol(Protocol):
    a2a_ids: tuple[str, ...]
    message: str
    is_a2a: bool          # 这次回复是不是一个"委托指令"

# :81 消息片段（文本 + 可选 JSON schema 元数据）
class PartsDict(TypedDict):
    text: str
    metadata: NotRequired[PartsMetadataDict]

# :99 更新机制注册表：轮询/流式/推送三选一
HANDLER_REGISTRY: dict[type[UpdateConfig], HandlerType] = {
    PollingConfig: PollingHandler,
    StreamingConfig: StreamingHandler,
    PushNotificationConfig: PushNotificationHandler,
}

描述一个远端 Agent 的配置（a2a/config.py:370）：

# a2a/config.py:370
class A2AConfig(BaseModel):
    """Deprecated: Use A2AClientConfig instead."""
    model_config = ConfigDict(extra="forbid")          # 多写一个字段就报错（防手滑）

    endpoint: Url = Field(description="A2A agent endpoint URL")   # 远端地址
    auth: ClientAuthScheme | None = Field(default=None, description="Authentication scheme")
    timeout: int = Field(default=120, description="Request timeout in seconds")
    max_turns: int = Field(default=10, description="Maximum conversation turns with A2A agent")
    response_model: type[BaseModel] | None = Field(default=None,
        description="Optional Pydantic model for structured A2A agent responses")
    fail_fast: bool = Field(default=True,
        description="If True, raise error when agent unreachable; if False, skip")
    trust_remote_completion_status: bool = Field(default=False,
        description="If True, return A2A result directly when completed")
    updates: UpdateConfig = Field(default_factory=_get_default_update_config)
    transport: ClientTransportConfig = Field(default_factory=ClientTransportConfig)

能力是"包"上去的，不改 Agent 类本身（a2a/wrapper.py:97）：

# a2a/wrapper.py:97
def wrap_agent_with_a2a_instance(agent, extension_registry=None):
    if extension_registry is None:
        extension_registry = ExtensionRegistry()
    extension_registry.inject_all_tools(agent)               # :113 注入 A2A 相关工具

    original_execute_task = agent.execute_task.__func__      # 存下原方法

    @wraps(original_execute_task)
    def execute_task_with_a2a(self, task, context=None, tools=None) -> str:
        if not self.a2a:                                     # 没配 a2a → 走原逻辑
            return original_execute_task(self, task, context, tools)
        a2a_agents, agent_response_model = get_a2a_agents_and_response_model(self.a2a)
        return _execute_task_with_a2a(self=self, a2a_agents=a2a_agents,
            original_fn=original_execute_task, task=task, ...)   # 有 a2a → 走委托逻辑

    object.__setattr__(agent, "execute_task", MethodType(execute_task_with_a2a, agent))  # :166 换掉
    ...  # kickoff / aexecute_task 同样被包裹

真正的委托发生在 _delegate_to_a2a 的 for 循环里（a2a/wrapper.py:1231）：

# a2a/wrapper.py:1261
ctx = _prepare_delegation_context(self, agent_response, task, original_task_description)
state = _init_delegation_state(ctx, agent_cards)
current_request = state.current_request
context_id = state.context_id; task_id = state.task_id
conversation_history = state.conversation_history

for turn_num in range(ctx.max_turns):                       # ★最多 max_turns 轮
    agent_branch, accepted_output_modes = _get_turn_context(ctx.agent_config)
    a2a_result = execute_a2a_delegation(                     # 发一次请求给远端
        endpoint=ctx.agent_config.endpoint,
        task_description=current_request,
        context_id=context_id, task_id=task_id,             # 带上会话/任务 id 维持连续性
        conversation_history=conversation_history,
        turn_number=turn_num + 1,
        transport=ctx.agent_config.transport, ...)
    conversation_history = a2a_result.get("history", [])    # 更新对话历史
    if conversation_history:                                # 记住远端分配的 id
        latest = conversation_history[-1]
        if latest.task_id is not None:    task_id = latest.task_id
        if latest.context_id is not None: context_id = latest.context_id
    if a2a_result["status"] in [TaskState.completed, TaskState.input_required]:
        ...  # 完成 或 需要更多输入 → 收尾/继续（见 L06）

委托把"不变的配置"和"每轮变的状态"分成两个 NamedTuple（a2a/wrapper.py:61）：

# a2a/wrapper.py:61
class DelegationContext(NamedTuple):
    """Context prepared for A2A delegation.（整个委托过程不变）"""
    a2a_agents: list[A2AConfig | A2AClientConfig]
    agent_response_model: type[BaseModel] | None
    current_request: str
    agent_id: str
    agent_config: A2AConfig | A2AClientConfig
    context_id: str | None; task_id: str | None
    max_turns: int

# :81
class DelegationState(NamedTuple):
    """Mutable state for A2A delegation loop.（每轮可能变）"""
    current_request: str
    context_id: str | None; task_id: str | None
    reference_task_ids: list[str]
    conversation_history: list[Message]
    agent_card: AgentCard | None
    agent_name: str | None

转满 max_turns 还没完成，有专门的收尾（a2a/wrapper.py:761）：

# a2a/wrapper.py:761
def _handle_max_turns_exceeded(conversation_history, max_turns, ...) -> str:
    """Handle the case when max turns is exceeded."""
    if conversation_history:
        for msg in reversed(conversation_history):        # 从最新往回找
            if msg.role == Role.agent:
                text_parts = [part.root.text for part in msg.parts if part.root.kind == "text"]
                final_message = " ".join(text_parts) if text_parts else "Conversation completed"
                crewai_event_bus.emit(None, ...)          # 发事件（Day 26 审计）
                # ... 返回目前拿到的最好结果，而不是直接崩

P=lib/crewai/src/crewai/a2a
sed -n '37,104p'  $P/types.py       # 协议类型/版本/传输/更新机制
sed -n '370,434p' $P/config.py      # A2AConfig 字段（合同条款）
sed -n '97,169p'  $P/wrapper.py     # wrap_agent_with_a2a_instance 包裹
sed -n '1261,1320p' $P/wrapper.py   # _delegate_to_a2a 多轮循环
sed -n '61,95p'   $P/wrapper.py     # DelegationContext / DelegationState
ls $P/updates $P/extensions         # 轮询/流式/推送 + 扩展