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 task

4.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:

1. The HTTP router receives and validates the request

2. Format detection logic identifies message structure (standard A2A or Google A2A)

3. The message is converted to the appropriate internal format

4. The DeepCoreA2AAgent processes the message using the agent service

5. The response is formatted according to the detected input format

6. The HTTP router returns the response

5.2 Task Management

Tasks in DeepCore's A2A implementation:

1. Are uniquely identified with UUIDs

2. Have a defined lifecycle (created, running, completed, failed, canceled)

3. Are persisted in Redis for reliability

4. 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 endpoint

• GET /A2A/{agent_id}/agent.json - Agent card endpoint

• POST /A2A/{agent_id}/ - Message handling endpoint

• POST /A2A/{agent_id}/stream - Streaming endpoint

• POST /A2A/{agent_id}/tasks/send - Task submission endpoint

• POST /A2A/{agent_id}/tasks/get - Task status endpoint

• POST /A2A/{agent_id}/tasks/cancel - Task cancellation endpoint

• POST /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.