call_agent
Send prompts to A2A protocol agents through Claude Desktop to access extended capabilities and agent interactions.
Instructions
Call an agent with a prompt.
Args:
agent_name: Name of the agent to call
prompt: Prompt to send to the agent
Returns:
Dict with response from the agent
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| agent_name | Yes | ||
| prompt | Yes |
Implementation Reference
- a2a_mcp_server/server.py:190-310 (handler)The core handler function for the 'call_agent' tool. It verifies the agent in the registry, fetches the agent card if necessary, connects to the A2A server using A2aMinClient, sends the prompt as a message, and returns the task response including artifacts.@mcp.tool() async def call_agent(agent_name: str, prompt: str) -> Dict[str, Any]: """ Call an agent with a prompt. Args: agent_name: Name of the agent to call prompt: Prompt to send to the agent Returns: Dict with response from the agent """ try: logger.debug(f"call_agent called with agent_name={agent_name}, prompt={prompt[:50]}...") logger.debug(f"Current registry: {state.registry}") # Verify the agent exists if agent_name not in state.registry: logger.warning(f"Agent '{agent_name}' not found in registry") return { "status": "error", "message": f"Agent '{agent_name}' not found in registry" } # Get the URL for the agent url = state.registry[agent_name] logger.debug(f"Using URL: {url}") try: # Get or create agent card if needed if agent_name not in state.cache: logger.debug(f"Agent card not in cache, fetching for {agent_name}") card = await fetch_agent_card(url) if card: state.cache[agent_name] = card else: logger.error(f"Failed to fetch agent card for {agent_name}") return { "status": "error", "message": f"Failed to fetch agent card for {agent_name}" } # Create a client using A2aMinClient try: # Connect to the A2A server logger.debug(f"Connecting to A2A server at {url}") client = A2aMinClient.connect(url) # Prepare message and send logger.debug(f"Sending message to agent {agent_name}: {prompt[:50]}...") task_id = str(uuid.uuid4()) session_id = str(uuid.uuid4()) # Send the message and get response task = await client.send_message( message=prompt, task_id=task_id, session_id=session_id ) # Process and return the response logger.debug(f"Received task response: {task}") response_data = { "status": "success", "task_id": task.id, "state": task.status.state if hasattr(task, 'status') else "unknown", "artifacts": [] } # Extract artifacts artifacts = task.artifacts if hasattr(task, 'artifacts') else [] if artifacts: logger.debug(f"Received {len(artifacts)} artifacts") for artifact in artifacts: artifact_data = { "name": artifact.name, "content": [] } for part in artifact.parts: part_type = part.type if part_type == "text": artifact_data["content"].append({ "type": "text", "text": part.text }) elif part_type == "file": file_data = part.file artifact_data["content"].append({ "type": "file", "name": file_data.name, "mime_type": file_data.mimeType }) elif part_type == "data": artifact_data["content"].append({ "type": "data", "data": part.data }) response_data["artifacts"].append(artifact_data) logger.info(f"Successfully called agent {agent_name}") return response_data except Exception as e: logger.exception(f"Error using A2aMinClient: {e}") return { "status": "error", "message": f"Error calling agent with A2aMinClient: {str(e)}" } except Exception as e: logger.exception(f"Error calling agent {agent_name}: {e}") return { "status": "error", "message": f"Error calling agent: {str(e)}" } except Exception as e: logger.exception(f"Error in call_agent: {str(e)}") return {"status": "error", "message": f"Error: {str(e)}"}
- a2a_mcp_server/server.py:192-201 (schema)Input schema defined in the tool's docstring: agent_name (str), prompt (str). Output is Dict[str, Any] with status, task_id, state, artifacts.""" Call an agent with a prompt. Args: agent_name: Name of the agent to call prompt: Prompt to send to the agent Returns: Dict with response from the agent """
- a2a_mcp_server/server.py:190-190 (registration)The @mcp.tool() decorator registers the call_agent function as an MCP tool.@mcp.tool()
- a2a_mcp_server/server.py:46-64 (helper)Helper function used by call_agent to fetch and cache the agent's card if not already cached.async def fetch_agent_card(url: str) -> Optional[AgentCard]: """Fetch agent card from an A2A server using A2ACardResolver.""" try: logger.debug(f"Fetching agent card from {url}") # Use the A2ACardResolver from a2a_min to get the agent card card_resolver = A2ACardResolver(url) try: card = card_resolver.get_agent_card() logger.debug(f"Received agent card: {card}") return card except Exception as e: logger.error(f"Failed to fetch agent card: {str(e)}") return None except Exception as e: logger.exception(f"Error fetching agent card from {url}: {e}") return None