github/crewAI
1
1
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-01-10 06:57:57 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix: Allow A2A agents to be identified by skill ID in addition to endpoint URL

Browse Source 
This commit fixes issue #3897 where the LLM would return a skill.id
(e.g., 'Research') instead of the full endpoint URL, causing a
Pydantic validation error.

Changes:
- Added resolve_agent_identifier() function to map skill IDs to endpoints
- Added extract_agent_identifiers_from_cards() to collect both endpoints and skill IDs
- Modified _execute_task_with_a2a() to rebuild AgentResponse model after fetching AgentCards
- Updated _delegate_to_a2a() to use resolver for identifier resolution
- Updated _augment_prompt_with_a2a() to explicitly instruct LLM about both identifier types
- Added comprehensive unit tests for resolve_agent_identifier()
- Added integration tests replicating the exact issue from #3897

The fix allows the dynamic Pydantic model to accept both endpoint URLs
and skill IDs in the Literal constraint, then resolves skill IDs to
their canonical endpoints before delegation. This maintains backward
compatibility while fixing the validation error.

Fixes #3897

Co-Authored-By: João <joao@crewai.com>
This commit is contained in:
Devin AI
2025-11-12 17:51:48 +00:00
parent c205d2e8de
commit d141078e72
4 changed files with 550 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
lib/crewai/src/crewai/a2a/utils.py
View File

@@ -753,3 +753,99 @@ def get_a2a_agents_and_response_model(
"""
a2a_agents, agent_ids = extract_a2a_agent_ids_from_config(a2a_config=a2a_config)
return a2a_agents, create_agent_response_model(agent_ids)
def extract_agent_identifiers_from_cards(
a2a_agents: list[A2AConfig],
agent_cards: dict[str, AgentCard],
) -> tuple[str, ...]:
"""Extract all valid agent identifiers (endpoints and skill IDs) from agent cards.
Args:
a2a_agents: List of A2A agent configurations
agent_cards: Dictionary mapping endpoints to AgentCards
Returns:
Tuple of all valid identifiers (endpoints + skill IDs)
"""
identifiers = set()
for config in a2a_agents:
identifiers.add(config.endpoint)
for card in agent_cards.values():
if card.skills:
for skill in card.skills:
identifiers.add(skill.id)
return tuple(sorted(identifiers))
def resolve_agent_identifier(
identifier: str,
a2a_agents: list[A2AConfig],
agent_cards: dict[str, AgentCard],
) -> str:
"""Resolve an agent identifier (endpoint or skill ID) to a canonical endpoint.
This function allows both endpoint URLs and skill IDs to be used as agent identifiers.
If the identifier is already an endpoint, it's returned as-is. If it's a skill ID,
it's resolved to the endpoint of the agent card that contains that skill.
Args:
identifier: Either an endpoint URL or a skill ID
a2a_agents: List of A2A agent configurations
agent_cards: Dictionary mapping endpoints to AgentCards
Returns:
The canonical endpoint URL
Raises:
ValueError: If the identifier is unknown or ambiguous (matches multiple agents)
Examples:
>>> # Endpoint passthrough
>>> resolve_agent_identifier(
... "http://localhost:10001/.well-known/agent-card.json",
... a2a_agents,
... agent_cards
... )
'http://localhost:10001/.well-known/agent-card.json'
>>> # Skill ID resolution
>>> resolve_agent_identifier("Research", a2a_agents, agent_cards)
'http://localhost:10001/.well-known/agent-card.json'
"""
endpoints = {config.endpoint for config in a2a_agents}
if identifier in endpoints:
return identifier
matching_endpoints: list[str] = []
for endpoint, card in agent_cards.items():
if card.skills:
for skill in card.skills:
if skill.id == identifier:
matching_endpoints.append(endpoint)
break
if len(matching_endpoints) == 0:
available_endpoints = ", ".join(sorted(endpoints))
available_skill_ids = []
for card in agent_cards.values():
if card.skills:
available_skill_ids.extend([skill.id for skill in card.skills])
available_skills = ", ".join(sorted(set(available_skill_ids))) if available_skill_ids else "none"
raise ValueError(
f"Unknown A2A agent identifier '{identifier}'. "
f"Available endpoints: {available_endpoints}. "
f"Available skill IDs: {available_skills}."
)
if len(matching_endpoints) > 1:
endpoints_list = ", ".join(sorted(matching_endpoints))
raise ValueError(
f"Ambiguous skill ID '{identifier}' found in multiple agents: {endpoints_list}. "
f"Please use the specific endpoint URL to disambiguate."
)
return matching_endpoints[0]

33
lib/crewai/src/crewai/a2a/wrapper.py
View File

@@ -23,9 +23,12 @@ from crewai.a2a.templates import (
)
from crewai.a2a.types import AgentResponseProtocol
from crewai.a2a.utils import (
create_agent_response_model,
execute_a2a_delegation,
extract_agent_identifiers_from_cards,
fetch_agent_card,
get_a2a_agents_and_response_model,
resolve_agent_identifier,
)
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.a2a_events import (
@@ -190,6 +193,9 @@ def _execute_task_with_a2a(
finally:
task.description = original_description
agent_identifiers = extract_agent_identifiers_from_cards(a2a_agents, agent_cards)
agent_response_model = create_agent_response_model(agent_identifiers)
task.description = _augment_prompt_with_a2a(
a2a_agents=a2a_agents,
task_description=original_description,
@@ -301,6 +307,13 @@ def _augment_prompt_with_a2a(
IMPORTANT: You have the ability to delegate this task to remote A2A agents.
{agents_text}
AGENT IDENTIFICATION: When setting a2a_ids, you may use either:
1. The agent's endpoint URL (e.g., "http://localhost:10001/.well-known/agent-card.json")
2. The exact skill.id from the agent's skills list (e.g., "Research")
Prefer using endpoint URLs when possible to avoid ambiguity. If a skill.id appears on multiple agents, you MUST use the endpoint URL to specify which agent you want.
{history_text}{turn_info}
@@ -445,16 +458,20 @@ def _delegate_to_a2a(
ImportError: If a2a-sdk is not installed
"""
a2a_agents, agent_response_model = get_a2a_agents_and_response_model(self.a2a)
agent_ids = tuple(config.endpoint for config in a2a_agents)
current_request = str(agent_response.message)
agent_id = agent_response.a2a_ids[0]
agent_identifier = agent_response.a2a_ids[0]
if agent_id not in agent_ids:
raise ValueError(
f"Unknown A2A agent ID(s): {agent_response.a2a_ids} not in {agent_ids}"
agent_cards_dict = agent_cards or {}
try:
agent_endpoint = resolve_agent_identifier(
agent_identifier, a2a_agents, agent_cards_dict
)
except ValueError as e:
raise ValueError(
f"Failed to resolve A2A agent identifier '{agent_identifier}': {e}"
) from e
agent_config = next(filter(lambda x: x.endpoint == agent_id, a2a_agents))
agent_config = next(filter(lambda x: x.endpoint == agent_endpoint, a2a_agents))
task_config = task.config or {}
context_id = task_config.get("context_id")
task_id_config = task_config.get("task_id")
@@ -488,7 +505,7 @@ def _delegate_to_a2a(
metadata=metadata,
extensions=extensions,
conversation_history=conversation_history,
agent_id=agent_id,
agent_id=agent_endpoint,
agent_role=Role.user,
agent_branch=agent_branch,
response_model=agent_config.response_model,
@@ -501,7 +518,7 @@ def _delegate_to_a2a(
final_result, next_request = _handle_agent_response_and_continue(
self=self,
a2a_result=a2a_result,
agent_id=agent_id,
agent_id=agent_endpoint,
agent_cards=agent_cards,
a2a_agents=a2a_agents,
original_task_description=original_task_description,

245
lib/crewai/tests/a2a/test_resolve_agent_identifier.py Normal file
View File

@@ -0,0 +1,245 @@
"""Test resolve_agent_identifier function for A2A skill ID resolution."""
import pytest
from a2a.types import AgentCapabilities, AgentCard, AgentSkill
from crewai.a2a.config import A2AConfig
from crewai.a2a.utils import resolve_agent_identifier
@pytest.fixture
def sample_agent_configs():
"""Create sample A2A agent configurations."""
return [
A2AConfig(endpoint="http://localhost:10001/.well-known/agent-card.json"),
A2AConfig(endpoint="http://localhost:10002/.well-known/agent-card.json"),
]
@pytest.fixture
def sample_agent_cards():
"""Create sample AgentCards with skills."""
card1 = AgentCard(
name="Research Agent",
description="An expert research agent",
url="http://localhost:10001",
version="1.0.0",
capabilities=AgentCapabilities(),
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
skills=[
AgentSkill(
id="Research",
name="Research",
description="Conduct comprehensive research",
tags=["research", "analysis"],
examples=["Research quantum computing"],
)
],
)
card2 = AgentCard(
name="Writing Agent",
description="An expert writing agent",
url="http://localhost:10002",
version="1.0.0",
capabilities=AgentCapabilities(),
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
skills=[
AgentSkill(
id="Writing",
name="Writing",
description="Write high-quality content",
tags=["writing", "content"],
examples=["Write a blog post"],
)
],
)
return {
"http://localhost:10001/.well-known/agent-card.json": card1,
"http://localhost:10002/.well-known/agent-card.json": card2,
}
def test_resolve_endpoint_passthrough(sample_agent_configs, sample_agent_cards):
"""Test that endpoint URLs are returned as-is."""
endpoint = "http://localhost:10001/.well-known/agent-card.json"
result = resolve_agent_identifier(endpoint, sample_agent_configs, sample_agent_cards)
assert result == endpoint
def test_resolve_unique_skill_id(sample_agent_configs, sample_agent_cards):
"""Test that a unique skill ID resolves to the correct endpoint."""
result = resolve_agent_identifier("Research", sample_agent_configs, sample_agent_cards)
assert result == "http://localhost:10001/.well-known/agent-card.json"
result = resolve_agent_identifier("Writing", sample_agent_configs, sample_agent_cards)
assert result == "http://localhost:10002/.well-known/agent-card.json"
def test_resolve_unknown_identifier(sample_agent_configs, sample_agent_cards):
"""Test that unknown identifiers raise a descriptive error."""
with pytest.raises(ValueError) as exc_info:
resolve_agent_identifier("UnknownSkill", sample_agent_configs, sample_agent_cards)
error_msg = str(exc_info.value)
assert "Unknown A2A agent identifier 'UnknownSkill'" in error_msg
assert "Available endpoints:" in error_msg
assert "Available skill IDs:" in error_msg
assert "Research" in error_msg
assert "Writing" in error_msg
def test_resolve_ambiguous_skill_id():
"""Test that ambiguous skill IDs raise a descriptive error."""
configs = [
A2AConfig(endpoint="http://localhost:10001/.well-known/agent-card.json"),
A2AConfig(endpoint="http://localhost:10002/.well-known/agent-card.json"),
]
card1 = AgentCard(
name="Research Agent 1",
description="First research agent",
url="http://localhost:10001",
version="1.0.0",
capabilities=AgentCapabilities(),
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
skills=[
AgentSkill(
id="Research",
name="Research",
description="Conduct research",
tags=["research"],
examples=["Research topic"],
)
],
)
card2 = AgentCard(
name="Research Agent 2",
description="Second research agent",
url="http://localhost:10002",
version="1.0.0",
capabilities=AgentCapabilities(),
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
skills=[
AgentSkill(
id="Research",
name="Research",
description="Conduct research",
tags=["research"],
examples=["Research topic"],
)
],
)
cards = {
"http://localhost:10001/.well-known/agent-card.json": card1,
"http://localhost:10002/.well-known/agent-card.json": card2,
}
with pytest.raises(ValueError) as exc_info:
resolve_agent_identifier("Research", configs, cards)
error_msg = str(exc_info.value)
assert "Ambiguous skill ID 'Research'" in error_msg
assert "found in multiple agents" in error_msg
assert "http://localhost:10001/.well-known/agent-card.json" in error_msg
assert "http://localhost:10002/.well-known/agent-card.json" in error_msg
assert "Please use the specific endpoint URL to disambiguate" in error_msg
def test_resolve_with_no_skills():
"""Test resolution when agent cards have no skills."""
configs = [
A2AConfig(endpoint="http://localhost:10001/.well-known/agent-card.json"),
]
card = AgentCard(
name="Agent Without Skills",
description="An agent without skills",
url="http://localhost:10001",
version="1.0.0",
capabilities=AgentCapabilities(),
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
skills=[],
)
cards = {
"http://localhost:10001/.well-known/agent-card.json": card,
}
result = resolve_agent_identifier(
"http://localhost:10001/.well-known/agent-card.json", configs, cards
)
assert result == "http://localhost:10001/.well-known/agent-card.json"
with pytest.raises(ValueError) as exc_info:
resolve_agent_identifier("SomeSkill", configs, cards)
error_msg = str(exc_info.value)
assert "Unknown A2A agent identifier 'SomeSkill'" in error_msg
assert "Available skill IDs: none" in error_msg
def test_resolve_with_multiple_skills_same_card(sample_agent_configs):
"""Test resolution when a card has multiple skills."""
card = AgentCard(
name="Multi-Skill Agent",
description="An agent with multiple skills",
url="http://localhost:10001",
version="1.0.0",
capabilities=AgentCapabilities(),
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
skills=[
AgentSkill(
id="Research",
name="Research",
description="Conduct research",
tags=["research"],
examples=["Research topic"],
),
AgentSkill(
id="Analysis",
name="Analysis",
description="Analyze data",
tags=["analysis"],
examples=["Analyze data"],
),
],
)
cards = {
"http://localhost:10001/.well-known/agent-card.json": card,
}
result1 = resolve_agent_identifier("Research", sample_agent_configs[:1], cards)
assert result1 == "http://localhost:10001/.well-known/agent-card.json"
result2 = resolve_agent_identifier("Analysis", sample_agent_configs[:1], cards)
assert result2 == "http://localhost:10001/.well-known/agent-card.json"
def test_resolve_empty_agent_cards():
"""Test resolution with empty agent cards dictionary."""
configs = [
A2AConfig(endpoint="http://localhost:10001/.well-known/agent-card.json"),
]
cards = {}
result = resolve_agent_identifier(
"http://localhost:10001/.well-known/agent-card.json", configs, cards
)
assert result == "http://localhost:10001/.well-known/agent-card.json"
with pytest.raises(ValueError) as exc_info:
resolve_agent_identifier("SomeSkill", configs, cards)
error_msg = str(exc_info.value)
assert "Unknown A2A agent identifier 'SomeSkill'" in error_msg

184
lib/crewai/tests/a2a/test_skill_id_integration.py Normal file
View File

@@ -0,0 +1,184 @@
"""Integration test for A2A skill ID resolution (issue #3897)."""
import pytest
from a2a.types import AgentCapabilities, AgentCard, AgentSkill
from pydantic import BaseModel
from crewai.a2a.config import A2AConfig
from crewai.a2a.utils import (
create_agent_response_model,
extract_agent_identifiers_from_cards,
resolve_agent_identifier,
)
def test_skill_id_resolution_integration():
"""Test the complete flow of skill ID resolution as described in issue #3897.
This test replicates the exact scenario from the bug report:
1. User creates A2A config with endpoint URL
2. Remote agent has AgentCard with skill.id="Research"
3. LLM returns a2a_ids=["Research"] instead of the endpoint URL
4. System should resolve "Research" to the endpoint and proceed successfully
"""
a2a_config = A2AConfig(
endpoint="http://localhost:10001/.well-known/agent-card.json"
)
a2a_agents = [a2a_config]
agent_card = AgentCard(
name="Research Agent",
description="An expert research agent that can conduct thorough research",
url="http://localhost:10001",
version="1.0.0",
capabilities=AgentCapabilities(),
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
skills=[
AgentSkill(
id="Research",
name="Research",
description="Conduct comprehensive research on any topic",
tags=["research", "analysis", "information-gathering"],
examples=[
"Research the latest developments in quantum computing",
"What are the current trends in renewable energy?",
],
)
],
)
agent_cards = {
"http://localhost:10001/.well-known/agent-card.json": agent_card
}
identifiers = extract_agent_identifiers_from_cards(a2a_agents, agent_cards)
assert "http://localhost:10001/.well-known/agent-card.json" in identifiers
assert "Research" in identifiers
agent_response_model = create_agent_response_model(identifiers)
agent_response_data = {
"a2a_ids": ["Research"], # LLM uses skill ID instead of endpoint
"message": "Please research quantum computing developments",
"is_a2a": True,
}
agent_response = agent_response_model.model_validate(agent_response_data)
assert agent_response.a2a_ids == ("Research",)
assert agent_response.message == "Please research quantum computing developments"
assert agent_response.is_a2a is True
resolved_endpoint = resolve_agent_identifier(
"Research", a2a_agents, agent_cards
)
assert resolved_endpoint == "http://localhost:10001/.well-known/agent-card.json"
resolved_endpoint_direct = resolve_agent_identifier(
"http://localhost:10001/.well-known/agent-card.json",
a2a_agents,
agent_cards,
)
assert resolved_endpoint_direct == "http://localhost:10001/.well-known/agent-card.json"
def test_skill_id_validation_error_before_fix():
"""Test that demonstrates the original bug (for documentation purposes).
Before the fix, creating an AgentResponse model with only endpoints
would cause a validation error when the LLM returned a skill ID.
"""
endpoints_only = ("http://localhost:10001/.well-known/agent-card.json",)
agent_response_model_old = create_agent_response_model(endpoints_only)
agent_response_data = {
"a2a_ids": ["Research"],
"message": "Please research quantum computing",
"is_a2a": True,
}
with pytest.raises(Exception) as exc_info:
agent_response_model_old.model_validate(agent_response_data)
error_msg = str(exc_info.value)
assert "validation error" in error_msg.lower() or "literal" in error_msg.lower()
def test_multiple_agents_with_unique_skill_ids():
"""Test that multiple agents with unique skill IDs work correctly."""
a2a_agents = [
A2AConfig(endpoint="http://localhost:10001/.well-known/agent-card.json"),
A2AConfig(endpoint="http://localhost:10002/.well-known/agent-card.json"),
]
card1 = AgentCard(
name="Research Agent",
description="Research agent",
url="http://localhost:10001",
version="1.0.0",
capabilities=AgentCapabilities(),
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
skills=[
AgentSkill(
id="Research",
name="Research",
description="Conduct research",
tags=["research"],
)
],
)
card2 = AgentCard(
name="Writing Agent",
description="Writing agent",
url="http://localhost:10002",
version="1.0.0",
capabilities=AgentCapabilities(),
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
skills=[
AgentSkill(
id="Writing",
name="Writing",
description="Write content",
tags=["writing"],
)
],
)
agent_cards = {
"http://localhost:10001/.well-known/agent-card.json": card1,
"http://localhost:10002/.well-known/agent-card.json": card2,
}
identifiers = extract_agent_identifiers_from_cards(a2a_agents, agent_cards)
assert len(identifiers) == 4
assert "http://localhost:10001/.well-known/agent-card.json" in identifiers
assert "http://localhost:10002/.well-known/agent-card.json" in identifiers
assert "Research" in identifiers
assert "Writing" in identifiers
agent_response_model = create_agent_response_model(identifiers)
response1 = agent_response_model.model_validate({
"a2a_ids": ["Research"],
"message": "Do research",
"is_a2a": True,
})
assert response1.a2a_ids == ("Research",)
response2 = agent_response_model.model_validate({
"a2a_ids": ["Writing"],
"message": "Write content",
"is_a2a": True,
})
assert response2.a2a_ids == ("Writing",)
endpoint1 = resolve_agent_identifier("Research", a2a_agents, agent_cards)
assert endpoint1 == "http://localhost:10001/.well-known/agent-card.json"
endpoint2 = resolve_agent_identifier("Writing", a2a_agents, agent_cards)
assert endpoint2 == "http://localhost:10002/.well-known/agent-card.json"