DeepCore A2A
DeepCore A2A
Summary
This whitepaper presents DeepCore's implementation of the Agent-to-Agent (A2A) Protocol, an open standard designed to facilitate communication between independent AI agent systems. DeepCore's A2A integration enables seamless interaction between agents built on different frameworks, supporting streamlined task management, multi-format messaging, and real-time streaming capabilities.
1. Introduction
The AI agent ecosystem is rapidly evolving, with diverse agents built using different technologies and frameworks. The Agent-to-Agent (A2A) Protocol, developed by Google, addresses the critical need for standardized communication between these heterogeneous agent systems. DeepCore has implemented a comprehensive A2A solution that enables agents to discover capabilities, exchange complex information, and collaborate effectively.
This implementation is built upon the official Google A2A Protocol Specification, ensuring full compatibility with the standard and interoperability with other A2A-compatible systems. DeepCore extends the core protocol with additional capabilities while maintaining strict adherence to the specification's requirements for message formats, task management, and agent discovery mechanisms.
2. A2A Protocol Overview
The A2A Protocol provides a standardized approach for agent communication with the following core features:
Agent Discovery: Mechanisms for agents to discover each other's capabilities
Task Management: Protocols for task creation, monitoring, and lifecycle management
Message Exchange: Standards for transferring text, files, and structured data
Streaming Communication: Real-time event-based interaction patterns
Security: Standards for authentication and authorization
3. DeepCore A2A Architecture
DeepCore's A2A implementation follows a layered architecture:
┌─────────────────────────────────────┐
│ A2A HTTP Router │
├─────────────────────────────────────┤
│ DeepCore A2A Agent │
├─────────────────────────────────────┤
│ Agent Service Integration │
├─────────────────────────────────────┤
│ Redis Task Storage │
└─────────────────────────────────────┘3.1 Key Components
DeepCore A2A Agent: Adapts existing DeepCore agents to the A2A protocol
A2A HTTP Router: Handles HTTP endpoints for the A2A protocol operations
Agent Service Integration: Connects to existing DeepCore agent services
Redis-based Task Storage: Manages persistent task state
4. Core Features and Capabilities
4.1 Agent Discovery via Agent Cards
DeepCore agents expose their capabilities through A2A Agent Cards, allowing other agents to discover:
Basic agent information (name, description, version)
Supported capabilities (streaming, push notifications)
Available skills and tools
Authentication requirements
# Agent Card creation example
agent_card = AgentCard(
name=self.agent_data.get("name", "DeepCore Agent"),
description=self.agent_data.get("description", ""),
url=base_url,
version="1.0.0",
capabilities={
"streaming": True,
"pushNotifications": False,
"stateTransitionHistory": False,
"google_a2a_compatible": True,
"parts_array_format": True
},
skills=skills,
)4.2 Multi-Format Message Handling
DeepCore's A2A implementation supports multiple message formats:
Standard A2A protocol format
Google A2A format with parts-based messaging
Auto-format detection and conversion
4.3 Task Lifecycle Management
The implementation provides comprehensive task management:
Creation and tracking of tasks
Task state persistence via Redis
Task status monitoring
Task cancellation capabilities
# Task lifecycle example
async def handle_task(self, task):
# Process task
# ...
# Mark task as completed
task.status = TaskStatus(state=TaskState.COMPLETED)
# Store task state
self._save_task(task)
return task4.4 Real-time Streaming Communication
DeepCore's A2A implementation provides robust streaming capabilities using Server-Sent Events (SSE):
Streaming message responses
Task status updates
Progress notifications
Error handling
5. Implementation Details
5.1 Message Processing Flow
When a message is received through the A2A endpoint:
The HTTP router receives and validates the request
Format detection logic identifies message structure (standard A2A or Google A2A)
The message is converted to the appropriate internal format
The DeepCoreA2AAgent processes the message using the agent service
The response is formatted according to the detected input format
The HTTP router returns the response
5.2 Task Management
Tasks in DeepCore's A2A implementation:
Are uniquely identified with UUIDs
Have a defined lifecycle (created, running, completed, failed, canceled)
Are persisted in Redis for reliability
Can be monitored through polling or SSE streaming
5.3 Format Compatibility
DeepCore's implementation ensures compatibility through:
Format auto-detection
Message format conversion
Support for both standard A2A and Google A2A formats
Format preference persistence
6. API Endpoints
DeepCore's A2A implementation exposes the following key endpoints:
GET /A2A/{agent_id}/- Agent discovery endpointGET /A2A/{agent_id}/agent.json- Agent card endpointPOST /A2A/{agent_id}/- Message handling endpointPOST /A2A/{agent_id}/stream- Streaming endpointPOST /A2A/{agent_id}/tasks/send- Task submission endpointPOST /A2A/{agent_id}/tasks/get- Task status endpointPOST /A2A/{agent_id}/tasks/cancel- Task cancellation endpointPOST /A2A/{agent_id}/tasks/stream- Task streaming endpoint
7. Use Cases
DeepCore's A2A implementation enables various interaction patterns:
7.1 Simple Question-Answer
Clients can send simple queries to agents and receive direct responses.
7.2 Multi-turn Conversations
Agents can maintain conversation context for complex interactions requiring multiple exchanges.
7.3 Collaborative Task Execution
Multiple agents can work together by delegating subtasks and exchanging intermediate results.
7.4 Real-time Progress Monitoring
Clients can monitor long-running tasks through streaming updates.
8. Code Examples
DeepCore's A2A implementation can be easily integrated with the python-a2a library. Here are practical examples of how to interact with DeepCore agents using the library:
8.1 Basic Message Interaction
Send a simple message to a DeepCore agent and get a response:
import asyncio
from python_a2a import A2AClient, Message, TextContent, MessageRole
async def basic_message_example(agent_id, token):
# Initialize the client with the agent's endpoint
client = A2AClient(
f"https://deepcore.top/A2A/{agent_id}",
headers={"X-API-Token": f"{token}"}
)
# Create a message
message = Message(
content=TextContent(text="What is the Agent-to-Agent protocol?"),
role=MessageRole.USER
)
# Send the message and wait for response
try:
response = await client.send_message(message)
print(f"Agent response: {response.content.text}")
return response
except Exception as e:
print(f"Error communicating with agent: {e}")
return None
# Run the example
if __name__ == "__main__":
agent_id = "your_agent_id"
token = "your_api_token"
asyncio.run(basic_message_example(agent_id, token))8.2 Streaming Responses
Get real-time streaming responses from a DeepCore agent:
import asyncio
from python_a2a import StreamingClient, Message, TextContent, MessageRole
async def streaming_example(agent_id, token):
# Initialize streaming client
client = StreamingClient(
f"https://deepcore.top/A2A/{agent_id}",
headers={"X-API-Token": f"{token}"}
)
# Create a message
message = Message(
content=TextContent(text="Tell me about A2A streaming capabilities"),
role=MessageRole.USER
)
# Stream the response
try:
print("Streaming response:")
async for chunk in client.stream_response(message):
if 'content' in chunk:
print(chunk['content'], end="", flush=True)
# Check for last chunk
if chunk.get('lastChunk', False):
print("\nStream complete")
break
except Exception as e:
print(f"\nStreaming error: {e}")
# Run the example
if __name__ == "__main__":
agent_id = "your_agent_id"
token = "your_api_token"
asyncio.run(streaming_example(agent_id, token))8.3 Task Management
Create, monitor, and manage tasks:
import asyncio
import time
from python_a2a import A2AClient, Task, Message, TextContent, MessageRole
async def task_management_example(agent_id, token):
# Initialize the client
client = A2AClient(
f"https://deepcore.top/A2A/{agent_id}",
headers={"X-API-Token": f"{token}"}
)
# Create a message
message = Message(
content=TextContent(text="Generate a comprehensive report on renewable energy"),
role=MessageRole.USER
)
# Create a task
task_id = f"task-{int(time.time())}"
task = Task(
id=task_id,
message=message
)
try:
# Send the task
print(f"Submitting task {task_id}...")
submitted_task = await client.send_task(task)
print(f"Task submitted with ID: {submitted_task.id}")
print(f"Initial status: {submitted_task.status.state}")
# Poll for task completion
max_attempts = 10
for attempt in range(max_attempts):
# Wait before checking status
await asyncio.sleep(2)
# Check task status
task_status = await client.get_task(task_id)
print(f"Task status ({attempt+1}/{max_attempts}): {task_status.status.state}")
# If task is complete, show results
if task_status.status.state in ["completed", "failed", "canceled"]:
if task_status.artifacts:
print("\nTask results:")
for artifact in task_status.artifacts:
for part in artifact.get("parts", []):
if part.get("type") == "text":
print(f"- {part.get('text')}")
break
# Task can be canceled if needed
# await client.cancel_task(task_id)
except Exception as e:
print(f"Error in task management: {e}")
# Run the example
if __name__ == "__main__":
agent_id = "your_agent_id"
token = "your_api_token"
asyncio.run(task_management_example(agent_id, token))8.4 Agent Networks
Work with multiple DeepCore agents as a coordinated network:
import asyncio
from python_a2a import AgentNetwork, Message, TextContent, MessageRole
async def agent_network_example(token):
# Create an agent network
network = AgentNetwork(name="DeepCore Specialized Agents")
# Add multiple agents to the network
network.add("research",
f"https://deepcore.top/A2A/research_agent",
headers={"X-API-Token": f"{token}"})
network.add("coding",
f"https://deepcore.top/A2A/code_agent",
headers={"X-API-Token": f"{token}"})
network.add("data_analysis",
f"https://deepcore.top/A2A/data_agent",
headers={"X-API-Token": f"{token}"})
# List all agents in the network
print("Agents in network:")
for agent_info in network.list_agents():
print(f"- {agent_info['name']} at {agent_info['url']}")
# Send messages to specific agents based on the task type
try:
# Research query to research agent
research_agent = network.get_agent("research")
research_message = Message(
content=TextContent(text="What are the latest advancements in quantum computing?"),
role=MessageRole.USER
)
research_response = await research_agent.send_message(research_message)
print("\nResearch Agent Response:")
print(research_response.content.text[:300] + "...")
# Coding query to coding agent
code_agent = network.get_agent("coding")
code_message = Message(
content=TextContent(text="Write a Python function to calculate Fibonacci numbers"),
role=MessageRole.USER
)
code_response = await code_agent.send_message(code_message)
print("\nCoding Agent Response:")
print(code_response.content.text[:300] + "...")
except Exception as e:
print(f"Error in agent network communication: {e}")
# Run the example
if __name__ == "__main__":
token = "your_api_token"
asyncio.run(agent_network_example(token))8.5 Integration with External Tools
Connect DeepCore agents with external tools using the Model Context Protocol (MCP):
import asyncio
from python_a2a import A2AClient, Message, TextContent, MessageRole, FunctionCallContent, FunctionResponseContent
async def tool_integration_example(agent_id, token):
# Initialize the client
client = A2AClient(
f"https://deepcore.top/A2A/{agent_id}",
headers={"X-API-Token": f"{token}"}
)
# Create a message requesting weather information
message = Message(
content=TextContent(text="What's the weather like in New York?"),
role=MessageRole.USER
)
try:
# Send the message
response = await client.send_message(message)
# Check if the response is a function call (tool request)
if hasattr(response.content, 'type') and response.content.type == 'function_call':
function_name = response.content.name
parameters = response.content.parameters
print(f"Agent requested tool: {function_name}")
print(f"Parameters: {parameters}")
# Simulate calling an external weather API
if function_name == "get_weather":
# In a real implementation, you would call an actual weather API here
weather_data = {
"location": parameters.get("location", "New York"),
"temperature": 72,
"conditions": "Partly Cloudy",
"humidity": 65
}
# Send function response back to the agent
function_response = Message(
content=FunctionResponseContent(
name=function_name,
response=weather_data
),
role=MessageRole.FUNCTION
)
# Get final response after providing the tool result
final_response = await client.send_message(function_response)
print("\nFinal agent response after tool use:")
print(final_response.content.text)
else:
print(f"Unknown function: {function_name}")
else:
# Direct response without tool use
print("\nDirect agent response:")
print(response.content.text)
except Exception as e:
print(f"Error in tool integration: {e}")
# Run the example
if __name__ == "__main__":
agent_id = "your_agent_id"
token = "your_api_token"
asyncio.run(tool_integration_example(agent_id, token))8.6 DeepCore A2A Resources
For more information and to get started with DeepCore's A2A implementation, visit the following resources:
Main A2A Endpoint: https://deepcore.top/a2a
API Documentation: https://docs.deepcore.top/
Python A2A Library: https://github.com/themanojdesai/python-a2a
A2A Protocol Specification: https://google.github.io/A2A/specification/
9. Security Considerations
DeepCore's A2A implementation addresses several security aspects:
Transport security through HTTPS
Authentication through standard HTTP mechanisms
Authorization based on agent and user identity
Input validation to prevent injection attacks
Resource management to prevent abuse
10. Future Directions
The DeepCore A2A implementation roadmap includes:
Enhanced push notification support
Multi-agent orchestration capabilities
Advanced authentication mechanisms
Expanded file exchange capabilities
Structured data schema negotiation
11. Conclusion
DeepCore's A2A implementation provides a robust framework for agent-to-agent communication, enabling interoperability between diverse AI systems. By adhering to the A2A protocol standard while supporting flexible message formats, DeepCore enables seamless integration between heterogeneous agent ecosystems.
The implementation's support for task management, streaming communication, and format compatibility positions DeepCore as a versatile platform for building complex multi-agent systems that can effectively collaborate to achieve user goals.
Last updated