Release v2.3.38

README.md

@@ -1628,6 +1628,48 @@ server.run()  # stdio for Claude Desktop
 | Security | Origin validation, DNS rebinding prevention |
 | WebSocket | Auto-reconnect with exponential backoff |
 
+## A2A (Agent2Agent Protocol)
+
+PraisonAI supports the [A2A Protocol](https://a2a-protocol.org) for agent-to-agent communication, enabling your agents to be discovered and collaborate with other AI agents.
+
+### A2A Server (Expose Agent as A2A Server)
+```python
+from praisonaiagents import Agent, A2A
+from fastapi import FastAPI
+
+# Create an agent with tools
+def search_web(query: str) -> str:
+    """Search the web for information."""
+    return f"Results for: {query}"
+
+agent = Agent(
+    name="Research Assistant",
+    role="Research Analyst",
+    goal="Help users research topics",
+    tools=[search_web]
+)
+
+# Expose as A2A Server
+a2a = A2A(agent=agent, url="http://localhost:8000/a2a")
+
+app = FastAPI()
+app.include_router(a2a.get_router())
+
+# Run: uvicorn app:app --reload
+# Agent Card: GET /.well-known/agent.json
+# Status: GET /status
+```
+
+### A2A Features
+| Feature | Description |
+|---------|-------------|
+| Agent Card | JSON metadata for agent discovery |
+| Skills Extraction | Auto-generate skills from tools |
+| Task Management | Stateful task lifecycle |
+| Streaming | SSE streaming for real-time updates |
+
+> **Documentation**: [docs.praison.ai/a2a](https://docs.praison.ai/a2a) | **Examples**: [examples/python/a2a](https://github.com/MervinPraison/PraisonAI/tree/main/examples/python/a2a)
+
 ## CLI Features
 
 | Feature | Docs |

docker/Dockerfile.chat

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=2.3.37" \
+    "praisonai>=2.3.38" \
     "praisonai[chat]" \
     "embedchain[github,youtube]"
 

docker/Dockerfile.dev

@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=2.3.37" \
+    "praisonai>=2.3.38" \
     "praisonai[ui]" \
     "praisonai[chat]" \
     "praisonai[realtime]" \

docker/Dockerfile.ui

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=2.3.37" \
+    "praisonai>=2.3.38" \
     "praisonai[ui]" \
     "praisonai[crewai]"
 

examples/python/a2a/a2a-server.py

@@ -0,0 +1,53 @@
+"""
+PraisonAI A2A Server Example
+
+Expose a PraisonAI Agent as an A2A (Agent2Agent) Server.
+This enables other AI agents to discover and communicate with your agent.
+
+Run:
+    uvicorn a2a-server:app --reload
+
+Endpoints:
+    GET /.well-known/agent.json  - Agent Card for discovery
+    GET /status                   - Server status
+"""
+
+from praisonaiagents import Agent, A2A
+from fastapi import FastAPI
+
+# Create an agent with tools
+def search_web(query: str) -> str:
+    """Search the web for information."""
+    return f"Search results for: {query}"
+
+def calculate(expression: str) -> str:
+    """Calculate a mathematical expression."""
+    try:
+        return f"Result: {eval(expression)}"
+    except Exception:
+        return "Invalid expression"
+
+agent = Agent(
+    name="Research Assistant",
+    role="Research Analyst",
+    goal="Help users research topics and answer questions",
+    tools=[search_web, calculate]
+)
+
+# Expose as A2A Server
+a2a = A2A(
+    agent=agent,
+    url="http://localhost:8000/a2a",
+    version="1.0.0"
+)
+
+# Create FastAPI app
+app = FastAPI(
+    title="PraisonAI A2A Server",
+    description="A2A-compatible agent server"
+)
+app.include_router(a2a.get_router())
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

src/praisonai-agents/praisonaiagents/__init__.py

@@ -314,3 +314,12 @@ except ImportError:
     _agui_available = False
     AGUI = None
 
+# A2A support (optional, lazy loaded)
+try:
+    from praisonaiagents.ui.a2a import A2A
+    __all__.append('A2A')
+    _a2a_available = True
+except ImportError:
+    _a2a_available = False
+    A2A = None
+

src/praisonai-agents/praisonaiagents/ui/a2a/__init__.py

@@ -0,0 +1,74 @@
+"""
+A2A Protocol Integration for PraisonAI Agents
+
+This module provides A2A (Agent2Agent) protocol support,
+enabling PraisonAI Agents to be exposed as A2A Servers
+for agent-to-agent communication.
+
+Usage:
+    from praisonaiagents import Agent
+    from praisonaiagents.ui.a2a import A2A
+    from fastapi import FastAPI
+
+    agent = Agent(name="Assistant", role="Helper", goal="Help users")
+    a2a = A2A(agent=agent)
+
+    app = FastAPI()
+    app.include_router(a2a.get_router())
+    
+    # Agent Card available at: GET /.well-known/agent.json
+    # A2A endpoint at: POST /a2a (JSON-RPC)
+"""
+
+from praisonaiagents.ui.a2a.a2a import A2A
+from praisonaiagents.ui.a2a.types import (
+    # Enums
+    TaskState,
+    Role,
+    # Part types
+    TextPart,
+    FilePart,
+    DataPart,
+    # Core types
+    Message,
+    TaskStatus,
+    Task,
+    Artifact,
+    # Agent Card types
+    AgentSkill,
+    AgentCapabilities,
+    AgentCard,
+    # Request/Response types
+    MessageConfig,
+    SendMessageRequest,
+    # Streaming events
+    TaskStatusUpdateEvent,
+    TaskArtifactUpdateEvent,
+)
+
+__all__ = [
+    # Main class
+    "A2A",
+    # Enums
+    "TaskState",
+    "Role",
+    # Part types
+    "TextPart",
+    "FilePart",
+    "DataPart",
+    # Core types
+    "Message",
+    "TaskStatus",
+    "Task",
+    "Artifact",
+    # Agent Card types
+    "AgentSkill",
+    "AgentCapabilities",
+    "AgentCard",
+    # Request/Response types
+    "MessageConfig",
+    "SendMessageRequest",
+    # Streaming events
+    "TaskStatusUpdateEvent",
+    "TaskArtifactUpdateEvent",
+]

src/praisonai-agents/praisonaiagents/ui/a2a/a2a.py

@@ -0,0 +1,154 @@
+"""
+A2A - Main A2A Interface Class
+
+Exposes PraisonAI Agents via the A2A (Agent2Agent) protocol.
+"""
+
+import logging
+from typing import List, Optional, TYPE_CHECKING
+
+from praisonaiagents.ui.a2a.types import AgentCard
+from praisonaiagents.ui.a2a.agent_card import generate_agent_card
+from praisonaiagents.ui.a2a.task_store import TaskStore
+
+if TYPE_CHECKING:
+    from praisonaiagents import Agent, PraisonAIAgents
+    from fastapi import APIRouter
+
+logger = logging.getLogger(__name__)
+
+
+class A2A:
+    """
+    A2A Interface for PraisonAI Agents.
+    
+    Exposes a PraisonAI Agent or PraisonAIAgents workflow via the A2A protocol,
+    enabling agent-to-agent communication with other A2A-compatible systems.
+    
+    Usage:
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a import A2A
+        from fastapi import FastAPI
+        
+        agent = Agent(name="Assistant", role="Helper", goal="Help users")
+        a2a = A2A(agent=agent, url="http://localhost:8000/a2a")
+        
+        app = FastAPI()
+        app.include_router(a2a.get_router())
+        
+        # Agent Card at: GET /.well-known/agent.json
+        # A2A endpoint at: POST /a2a
+    
+    Args:
+        agent: Single PraisonAI Agent instance
+        agents: PraisonAIAgents instance for multi-agent workflows
+        name: Name for the A2A endpoint (defaults to agent name)
+        description: Description of the agent
+        url: URL where the A2A endpoint is hosted
+        version: Version string for the agent
+        prefix: URL prefix for the router
+        tags: OpenAPI tags for the router
+    """
+    
+    def __init__(
+        self,
+        agent: Optional["Agent"] = None,
+        agents: Optional["PraisonAIAgents"] = None,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+        url: str = "http://localhost:8000/a2a",
+        version: str = "1.0.0",
+        prefix: str = "",
+        tags: Optional[List[str]] = None,
+    ):
+        if agent is None and agents is None:
+            raise ValueError("A2A requires an agent or agents instance")
+        
+        self.agent = agent
+        self.agents = agents
+        self.url = url
+        self.version = version
+        self.prefix = prefix
+        self.tags = tags or ["A2A"]
+        
+        # Set name from agent if not provided
+        if name:
+            self.name = name
+        elif agent and hasattr(agent, 'name'):
+            self.name = agent.name
+        elif agents and hasattr(agents, 'name'):
+            self.name = agents.name
+        else:
+            self.name = "PraisonAI Agent"
+        
+        # Set description
+        if description:
+            self.description = description
+        elif agent and hasattr(agent, 'role'):
+            self.description = agent.role
+        else:
+            self.description = "PraisonAI Agent via A2A"
+        
+        # Initialize task store
+        self.task_store = TaskStore()
+        
+        # Router cache
+        self._router: Optional["APIRouter"] = None
+        self._agent_card: Optional[AgentCard] = None
+    
+    def get_agent_card(self) -> AgentCard:
+        """
+        Get the Agent Card for this A2A instance.
+        
+        Returns:
+            AgentCard object for A2A discovery
+        """
+        if self._agent_card is None:
+            self._agent_card = generate_agent_card(
+                agent=self.agent,
+                url=self.url,
+                version=self.version,
+                streaming=True,
+            )
+        return self._agent_card
+    
+    def get_router(self) -> "APIRouter":
+        """
+        Get the FastAPI router for this A2A instance.
+        
+        Returns:
+            FastAPI APIRouter with A2A endpoints
+        """
+        if self._router is None:
+            self._router = self._create_router()
+        return self._router
+    
+    def _create_router(self) -> "APIRouter":
+        """Create FastAPI router with A2A endpoints."""
+        from fastapi import APIRouter
+        from fastapi.responses import JSONResponse
+        
+        router = APIRouter(prefix=self.prefix, tags=self.tags)
+        
+        # Agent Card endpoint (well-known URI)
+        @router.get("/.well-known/agent.json")
+        async def get_agent_card():
+            """Return the Agent Card for discovery."""
+            card = self.get_agent_card()
+            return JSONResponse(content=card.model_dump(by_alias=True, exclude_none=True))
+        
+        # Status endpoint
+        @router.get("/status")
+        async def get_status():
+            """Return server status."""
+            return {
+                "status": "ok",
+                "name": self.name,
+                "version": self.version,
+            }
+        
+        # TODO: Add JSON-RPC endpoint for message/send
+        # TODO: Add streaming endpoint for message/stream
+        # TODO: Add task management endpoints
+        
+        return router

src/praisonai-agents/praisonaiagents/ui/a2a/agent_card.py

@@ -0,0 +1,123 @@
+"""
+A2A Agent Card Generation
+
+Generate Agent Cards from PraisonAI Agents for A2A discovery.
+"""
+
+import inspect
+from typing import TYPE_CHECKING, List, Optional
+
+from praisonaiagents.ui.a2a.types import (
+    AgentCard,
+    AgentCapabilities,
+    AgentSkill,
+)
+
+if TYPE_CHECKING:
+    from praisonaiagents import Agent
+
+
+def extract_skills_from_tools(tools: Optional[List]) -> List[AgentSkill]:
+    """
+    Extract AgentSkill objects from a list of tools.
+    
+    Args:
+        tools: List of tool functions or objects
+        
+    Returns:
+        List of AgentSkill objects
+    """
+    if not tools:
+        return []
+    
+    skills = []
+    
+    for tool in tools:
+        # Skip non-callable items
+        if not callable(tool):
+            continue
+        
+        # Get function name
+        name = getattr(tool, '__name__', None)
+        if not name:
+            continue
+        
+        # Get description from docstring
+        docstring = inspect.getdoc(tool)
+        description = docstring.split('\n')[0] if docstring else f"Tool: {name}"
+        
+        # Create skill
+        skill = AgentSkill(
+            id=name,
+            name=name,
+            description=description,
+            tags=["tool"],
+        )
+        skills.append(skill)
+    
+    return skills
+
+
+def generate_agent_card(
+    agent: "Agent",
+    url: str,
+    version: str = "1.0.0",
+    streaming: bool = False,
+    push_notifications: bool = False,
+) -> AgentCard:
+    """
+    Generate an A2A Agent Card from a PraisonAI Agent.
+    
+    Args:
+        agent: PraisonAI Agent instance
+        url: URL where the A2A endpoint is hosted
+        version: Version string for the agent
+        streaming: Whether streaming is supported
+        push_notifications: Whether push notifications are supported
+        
+    Returns:
+        AgentCard object for A2A discovery
+    """
+    # Get agent name
+    name = getattr(agent, 'name', None) or "PraisonAI Agent"
+    
+    # Build description from role, goal, or instructions
+    description_parts = []
+    
+    role = getattr(agent, 'role', None)
+    if role:
+        description_parts.append(role)
+    
+    goal = getattr(agent, 'goal', None)
+    if goal:
+        description_parts.append(goal)
+    
+    instructions = getattr(agent, 'instructions', None)
+    if instructions and not description_parts:
+        # Use first line of instructions if no role/goal
+        description_parts.append(instructions.split('\n')[0][:200])
+    
+    description = ". ".join(description_parts) if description_parts else "PraisonAI Agent"
+    
+    # Extract skills from tools
+    tools = getattr(agent, 'tools', None)
+    skills = extract_skills_from_tools(tools)
+    
+    # Build capabilities
+    capabilities = AgentCapabilities(
+        streaming=streaming,
+        push_notifications=push_notifications,
+    )
+    
+    # Create Agent Card
+    card = AgentCard(
+        name=name,
+        url=url,
+        version=version,
+        description=description,
+        capabilities=capabilities,
+        skills=skills if skills else None,
+        provider={"name": "PraisonAI"},
+    )
+    
+    return card

src/praisonai-agents/praisonaiagents/ui/a2a/conversion.py

@@ -0,0 +1,131 @@
+"""
+A2A Message Conversion
+
+Convert between A2A protocol messages and PraisonAI internal format.
+"""
+
+import uuid
+from typing import Any, Dict, List, Optional
+
+from praisonaiagents.ui.a2a.types import (
+    Message,
+    Role,
+    TextPart,
+    Artifact,
+)
+
+
+def a2a_to_praisonai_messages(messages: List[Message]) -> List[Dict[str, Any]]:
+    """
+    Convert A2A messages to PraisonAI/OpenAI chat format.
+    
+    Args:
+        messages: List of A2A Message objects
+        
+    Returns:
+        List of dicts in OpenAI chat format {"role": ..., "content": ...}
+    """
+    result = []
+    
+    for msg in messages:
+        # Map A2A role to OpenAI role
+        role = "user" if msg.role == Role.USER else "assistant"
+        
+        # Combine all text parts
+        content_parts = []
+        for part in msg.parts:
+            if isinstance(part, TextPart):
+                content_parts.append(part.text)
+            elif hasattr(part, 'text'):
+                content_parts.append(part.text)
+        
+        content = " ".join(content_parts)
+        
+        result.append({
+            "role": role,
+            "content": content,
+        })
+    
+    return result
+
+
+def praisonai_to_a2a_message(
+    response: str,
+    message_id: Optional[str] = None,
+    context_id: Optional[str] = None,
+    task_id: Optional[str] = None,
+) -> Message:
+    """
+    Convert PraisonAI response to A2A Message.
+    
+    Args:
+        response: Text response from agent
+        message_id: Optional message ID (auto-generated if not provided)
+        context_id: Optional context ID
+        task_id: Optional task ID
+        
+    Returns:
+        A2A Message object
+    """
+    msg_id = message_id or f"msg-{uuid.uuid4().hex[:12]}"
+    
+    return Message(
+        message_id=msg_id,
+        role=Role.AGENT,
+        parts=[TextPart(text=response)],
+        context_id=context_id,
+        task_id=task_id,
+    )
+
+
+def create_artifact(
+    content: str,
+    artifact_id: Optional[str] = None,
+    name: Optional[str] = None,
+    description: Optional[str] = None,
+) -> Artifact:
+    """
+    Create an A2A Artifact from response content.
+    
+    Args:
+        content: Text content for the artifact
+        artifact_id: Optional artifact ID (auto-generated if not provided)
+        name: Optional artifact name
+        description: Optional artifact description
+        
+    Returns:
+        A2A Artifact object
+    """
+    art_id = artifact_id or f"art-{uuid.uuid4().hex[:12]}"
+    
+    return Artifact(
+        artifact_id=art_id,
+        name=name,
+        description=description,
+        parts=[TextPart(text=content)],
+    )
+
+
+def extract_user_input(messages: List[Message]) -> str:
+    """
+    Extract the last user input from a list of messages.
+    
+    Args:
+        messages: List of A2A Message objects
+        
+    Returns:
+        Text content of the last user message, or empty string if none
+    """
+    # Find the last user message
+    for msg in reversed(messages):
+        if msg.role == Role.USER:
+            # Combine text parts
+            content_parts = []
+            for part in msg.parts:
+                if isinstance(part, TextPart):
+                    content_parts.append(part.text)
+                elif hasattr(part, 'text'):
+                    content_parts.append(part.text)
+            return " ".join(content_parts)
+    
+    return ""

src/praisonai-agents/praisonaiagents/ui/a2a/streaming.py

@@ -0,0 +1,189 @@
+"""
+A2A Streaming Events
+
+SSE streaming support for A2A task updates.
+"""
+
+import json
+from typing import AsyncIterator, Optional
+
+from praisonaiagents.ui.a2a.types import (
+    Task,
+    TaskState,
+    TaskStatus,
+    TaskStatusUpdateEvent,
+    TaskArtifactUpdateEvent,
+    Artifact,
+    TextPart,
+)
+
+
+class A2AEventEncoder:
+    """Encode A2A events for SSE streaming."""
+    
+    @staticmethod
+    def encode_event(event_type: str, data: dict) -> str:
+        """
+        Encode an event for SSE streaming.
+        
+        Args:
+            event_type: Type of event (e.g., "task.status", "task.artifact")
+            data: Event data as dict
+            
+        Returns:
+            SSE-formatted string
+        """
+        json_data = json.dumps(data, default=str)
+        return f"event: {event_type}\ndata: {json_data}\n\n"
+    
+    @staticmethod
+    def encode_task_status(event: TaskStatusUpdateEvent) -> str:
+        """Encode TaskStatusUpdateEvent for SSE."""
+        data = event.model_dump(by_alias=True, exclude_none=True)
+        return A2AEventEncoder.encode_event("task.status", data)
+    
+    @staticmethod
+    def encode_task_artifact(event: TaskArtifactUpdateEvent) -> str:
+        """Encode TaskArtifactUpdateEvent for SSE."""
+        data = event.model_dump(by_alias=True, exclude_none=True)
+        return A2AEventEncoder.encode_event("task.artifact", data)
+    
+    @staticmethod
+    def encode_task(task: Task) -> str:
+        """Encode Task object for SSE."""
+        data = task.model_dump(by_alias=True, exclude_none=True)
+        return A2AEventEncoder.encode_event("task", data)
+    
+    @staticmethod
+    def encode_done() -> str:
+        """Encode stream completion event."""
+        return "event: done\ndata: {}\n\n"
+
+
+def create_status_event(
+    task_id: str,
+    state: TaskState,
+    context_id: Optional[str] = None,
+    final: bool = False,
+) -> TaskStatusUpdateEvent:
+    """
+    Create a TaskStatusUpdateEvent.
+    
+    Args:
+        task_id: Task ID
+        state: New task state
+        context_id: Optional context ID
+        final: Whether this is the final event
+        
+    Returns:
+        TaskStatusUpdateEvent
+    """
+    return TaskStatusUpdateEvent(
+        task_id=task_id,
+        context_id=context_id,
+        status=TaskStatus(state=state),
+        final=final,
+    )
+
+
+def create_artifact_event(
+    task_id: str,
+    content: str,
+    artifact_id: Optional[str] = None,
+    context_id: Optional[str] = None,
+    append: bool = False,
+    last_chunk: bool = False,
+) -> TaskArtifactUpdateEvent:
+    """
+    Create a TaskArtifactUpdateEvent.
+    
+    Args:
+        task_id: Task ID
+        content: Artifact content
+        artifact_id: Optional artifact ID
+        context_id: Optional context ID
+        append: Whether to append to existing artifact
+        last_chunk: Whether this is the last chunk
+        
+    Returns:
+        TaskArtifactUpdateEvent
+    """
+    import uuid
+    art_id = artifact_id or f"art-{uuid.uuid4().hex[:12]}"
+    
+    artifact = Artifact(
+        artifact_id=art_id,
+        parts=[TextPart(text=content)],
+    )
+    
+    return TaskArtifactUpdateEvent(
+        task_id=task_id,
+        context_id=context_id,
+        artifact=artifact,
+        append=append,
+        last_chunk=last_chunk,
+    )
+
+
+async def stream_agent_response(
+    agent,
+    user_input: str,
+    task_id: str,
+    context_id: Optional[str] = None,
+) -> AsyncIterator[str]:
+    """
+    Stream agent response as A2A events.
+    
+    Args:
+        agent: PraisonAI Agent instance
+        user_input: User input text
+        task_id: Task ID
+        context_id: Optional context ID
+        
+    Yields:
+        SSE-formatted event strings
+    """
+    encoder = A2AEventEncoder()
+    
+    # Emit working status
+    working_event = create_status_event(
+        task_id=task_id,
+        state=TaskState.WORKING,
+        context_id=context_id,
+    )
+    yield encoder.encode_task_status(working_event)
+    
+    try:
+        # Get response from agent
+        response = agent.chat(user_input)
+        
+        # Emit artifact with response
+        artifact_event = create_artifact_event(
+            task_id=task_id,
+            content=str(response),
+            context_id=context_id,
+            last_chunk=True,
+        )
+        yield encoder.encode_task_artifact(artifact_event)
+        
+        # Emit completed status
+        completed_event = create_status_event(
+            task_id=task_id,
+            state=TaskState.COMPLETED,
+            context_id=context_id,
+            final=True,
+        )
+        yield encoder.encode_task_status(completed_event)
+        
+    except Exception:
+        # Emit failed status
+        failed_event = create_status_event(
+            task_id=task_id,
+            state=TaskState.FAILED,
+            context_id=context_id,
+            final=True,
+        )
+        yield encoder.encode_task_status(failed_event)
+    
+    # Emit done
+    yield encoder.encode_done()

src/praisonai-agents/praisonaiagents/ui/a2a/task_store.py

@@ -0,0 +1,209 @@
+"""
+A2A Task Store
+
+In-memory storage for A2A Task lifecycle management.
+"""
+
+import uuid
+from datetime import datetime, timezone
+from typing import Dict, List, Optional
+
+from praisonaiagents.ui.a2a.types import (
+    Task,
+    TaskState,
+    TaskStatus,
+    Message,
+    Artifact,
+)
+
+
+class TaskStore:
+    """
+    In-memory store for A2A Tasks.
+    
+    Manages task lifecycle including creation, status updates,
+    artifact storage, and message history.
+    """
+    
+    def __init__(self):
+        """Initialize empty task store."""
+        self._tasks: Dict[str, Task] = {}
+    
+    def create_task(self, message: Message, context_id: Optional[str] = None) -> Task:
+        """
+        Create a new task from an incoming message.
+        
+        Args:
+            message: The initial message that triggers the task
+            context_id: Optional context ID (uses message's context_id if not provided)
+            
+        Returns:
+            Newly created Task in SUBMITTED state
+        """
+        task_id = f"task-{uuid.uuid4().hex[:12]}"
+        ctx_id = context_id or message.context_id
+        
+        # Create initial status
+        status = TaskStatus(
+            state=TaskState.SUBMITTED,
+            timestamp=datetime.now(timezone.utc).isoformat(),
+        )
+        
+        # Create task with initial message in history
+        task = Task(
+            id=task_id,
+            context_id=ctx_id,
+            status=status,
+            history=[message],
+            artifacts=None,
+        )
+        
+        self._tasks[task_id] = task
+        return task
+    
+    def get_task(self, task_id: str) -> Optional[Task]:
+        """
+        Retrieve a task by its ID.
+        
+        Args:
+            task_id: The task ID to look up
+            
+        Returns:
+            Task if found, None otherwise
+        """
+        return self._tasks.get(task_id)
+    
+    def update_status(
+        self, 
+        task_id: str, 
+        state: TaskState,
+        message: Optional[Message] = None,
+    ) -> Optional[Task]:
+        """
+        Update the status of a task.
+        
+        Args:
+            task_id: The task ID to update
+            state: New task state
+            message: Optional status message
+            
+        Returns:
+            Updated Task if found, None otherwise
+        """
+        task = self._tasks.get(task_id)
+        if not task:
+            return None
+        
+        # Create new status
+        new_status = TaskStatus(
+            state=state,
+            message=message,
+            timestamp=datetime.now(timezone.utc).isoformat(),
+        )
+        
+        # Update task (create new instance for immutability)
+        updated = Task(
+            id=task.id,
+            context_id=task.context_id,
+            status=new_status,
+            artifacts=task.artifacts,
+            history=task.history,
+            metadata=task.metadata,
+        )
+        
+        self._tasks[task_id] = updated
+        return updated
+    
+    def add_artifact(self, task_id: str, artifact: Artifact) -> Optional[Task]:
+        """
+        Add an artifact to a task.
+        
+        Args:
+            task_id: The task ID to update
+            artifact: Artifact to add
+            
+        Returns:
+            Updated Task if found, None otherwise
+        """
+        task = self._tasks.get(task_id)
+        if not task:
+            return None
+        
+        # Get existing artifacts or create new list
+        artifacts = list(task.artifacts) if task.artifacts else []
+        artifacts.append(artifact)
+        
+        # Update task
+        updated = Task(
+            id=task.id,
+            context_id=task.context_id,
+            status=task.status,
+            artifacts=artifacts,
+            history=task.history,
+            metadata=task.metadata,
+        )
+        
+        self._tasks[task_id] = updated
+        return updated
+    
+    def add_to_history(self, task_id: str, message: Message) -> Optional[Task]:
+        """
+        Add a message to task history.
+        
+        Args:
+            task_id: The task ID to update
+            message: Message to add to history
+            
+        Returns:
+            Updated Task if found, None otherwise
+        """
+        task = self._tasks.get(task_id)
+        if not task:
+            return None
+        
+        # Get existing history or create new list
+        history = list(task.history) if task.history else []
+        history.append(message)
+        
+        # Update task
+        updated = Task(
+            id=task.id,
+            context_id=task.context_id,
+            status=task.status,
+            artifacts=task.artifacts,
+            history=history,
+            metadata=task.metadata,
+        )
+        
+        self._tasks[task_id] = updated
+        return updated
+    
+    def list_tasks(self, context_id: Optional[str] = None) -> List[Task]:
+        """
+        List all tasks, optionally filtered by context.
+        
+        Args:
+            context_id: Optional context ID to filter by
+            
+        Returns:
+            List of tasks
+        """
+        if context_id:
+            return [t for t in self._tasks.values() if t.context_id == context_id]
+        return list(self._tasks.values())
+    
+    def cancel_task(self, task_id: str) -> Optional[Task]:
+        """
+        Cancel a task.
+        
+        Args:
+            task_id: The task ID to cancel
+            
+        Returns:
+            Updated Task if found, None otherwise
+        """
+        return self.update_status(task_id, TaskState.CANCELLED)
+    
+    def clear(self):
+        """Clear all tasks from the store."""
+        self._tasks.clear()

src/praisonai-agents/praisonaiagents/ui/a2a/types.py

@@ -0,0 +1,266 @@
+"""
+A2A Protocol Types
+
+Pydantic models for the Agent2Agent (A2A) protocol.
+Based on the A2A Protocol Specification v1.0.
+"""
+
+from enum import Enum
+from typing import Any, Dict, List, Optional, Union
+from pydantic import BaseModel, ConfigDict, Field
+
+
+# =============================================================================
+# Enums
+# =============================================================================
+
+class TaskState(str, Enum):
+    """
+    Defines the possible lifecycle states of a Task.
+    
+    Based on A2A Protocol Specification Section 4.1.3.
+    """
+    SUBMITTED = "submitted"
+    WORKING = "working"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+    INPUT_REQUIRED = "input_required"
+    AUTH_REQUIRED = "auth_required"
+
+
+class Role(str, Enum):
+    """
+    Defines the sender of a message in A2A protocol communication.
+    
+    Based on A2A Protocol Specification Section 4.1.5.
+    """
+    USER = "user"
+    AGENT = "agent"
+
+
+# =============================================================================
+# Part Types
+# =============================================================================
+
+class TextPart(BaseModel):
+    """
+    A text content part.
+    
+    Based on A2A Protocol Specification Section 4.1.6.
+    """
+    text: str
+    metadata: Optional[Dict[str, Any]] = None
+
+
+class FilePart(BaseModel):
+    """
+    A file content part.
+    
+    Can contain either a URI reference or inline bytes.
+    Based on A2A Protocol Specification Section 4.1.7.
+    """
+    file_uri: Optional[str] = Field(default=None, alias="fileWithUri")
+    file_bytes: Optional[bytes] = Field(default=None, alias="fileWithBytes")
+    media_type: Optional[str] = Field(default=None, alias="mediaType")
+    name: Optional[str] = None
+    metadata: Optional[Dict[str, Any]] = None
+    
+    model_config = ConfigDict(populate_by_name=True)
+
+
+class DataPart(BaseModel):
+    """
+    A structured data part (JSON blob).
+    
+    Based on A2A Protocol Specification Section 4.1.8.
+    """
+    data: Dict[str, Any]
+    metadata: Optional[Dict[str, Any]] = None
+
+
+# Union type for all Part types
+Part = Union[TextPart, FilePart, DataPart]
+
+
+# =============================================================================
+# Core Types
+# =============================================================================
+
+class Message(BaseModel):
+    """
+    A communication turn between client and server.
+    
+    Based on A2A Protocol Specification Section 4.1.4.
+    """
+    model_config = ConfigDict(populate_by_name=True)
+    
+    message_id: str = Field(alias="messageId")
+    role: Role
+    parts: List[Part]
+    context_id: Optional[str] = Field(default=None, alias="contextId")
+    task_id: Optional[str] = Field(default=None, alias="taskId")
+    metadata: Optional[Dict[str, Any]] = None
+    extensions: Optional[Dict[str, str]] = None
+    reference_task_ids: Optional[List[str]] = Field(default=None, alias="referenceTaskIds")
+
+
+class TaskStatus(BaseModel):
+    """
+    Container for the status of a task.
+    
+    Based on A2A Protocol Specification Section 4.1.2.
+    """
+    state: TaskState
+    message: Optional[Message] = None
+    timestamp: Optional[str] = None
+
+
+class Artifact(BaseModel):
+    """
+    An output generated by the agent as a result of a task.
+    
+    Based on A2A Protocol Specification Section 4.1.9.
+    """
+    model_config = ConfigDict(populate_by_name=True)
+    
+    artifact_id: str = Field(alias="artifactId")
+    name: Optional[str] = None
+    description: Optional[str] = None
+    parts: List[Part]
+    metadata: Optional[Dict[str, Any]] = None
+    extensions: Optional[Dict[str, str]] = None
+
+
+class Task(BaseModel):
+    """
+    The core unit of action for A2A.
+    
+    Based on A2A Protocol Specification Section 4.1.1.
+    """
+    model_config = ConfigDict(populate_by_name=True)
+    
+    id: str
+    context_id: Optional[str] = Field(default=None, alias="contextId")
+    status: TaskStatus
+    artifacts: Optional[List[Artifact]] = None
+    history: Optional[List[Message]] = None
+    metadata: Optional[Dict[str, Any]] = None
+
+
+# =============================================================================
+# Agent Card Types
+# =============================================================================
+
+class AgentSkill(BaseModel):
+    """
+    Describes a skill/capability of an agent.
+    
+    Based on A2A Protocol Specification Section 4.4.
+    """
+    model_config = ConfigDict(populate_by_name=True)
+    
+    id: str
+    name: str
+    description: str
+    tags: Optional[List[str]] = None
+    examples: Optional[List[str]] = None
+    input_modes: Optional[List[str]] = Field(default=None, alias="inputModes")
+    output_modes: Optional[List[str]] = Field(default=None, alias="outputModes")
+
+
+class AgentCapabilities(BaseModel):
+    """
+    Describes the capabilities of an agent.
+    
+    Based on A2A Protocol Specification Section 4.4.
+    """
+    model_config = ConfigDict(populate_by_name=True)
+    
+    streaming: bool = False
+    push_notifications: bool = Field(default=False, alias="pushNotifications")
+    state_transition_history: bool = Field(default=False, alias="stateTransitionHistory")
+
+
+class AgentCard(BaseModel):
+    """
+    JSON metadata document describing an agent.
+    
+    Published by an A2A Server for discovery.
+    Based on A2A Protocol Specification Section 8.
+    """
+    model_config = ConfigDict(populate_by_name=True)
+    
+    name: str
+    url: str
+    version: str
+    description: Optional[str] = None
+    capabilities: AgentCapabilities
+    skills: Optional[List[AgentSkill]] = None
+    default_input_modes: Optional[List[str]] = Field(default=None, alias="defaultInputModes")
+    default_output_modes: Optional[List[str]] = Field(default=None, alias="defaultOutputModes")
+    provider: Optional[Dict[str, str]] = None
+    documentation_url: Optional[str] = Field(default=None, alias="documentationUrl")
+
+
+# =============================================================================
+# Request/Response Types
+# =============================================================================
+
+class MessageConfig(BaseModel):
+    """
+    Configuration for message sending.
+    
+    Based on A2A Protocol Specification Section 3.2.
+    """
+    model_config = ConfigDict(populate_by_name=True)
+    
+    accept_content_types: Optional[List[str]] = Field(default=None, alias="acceptContentTypes")
+    blocking: Optional[bool] = None
+    history_length: Optional[int] = Field(default=None, alias="historyLength")
+
+
+class SendMessageRequest(BaseModel):
+    """
+    Request to send a message to an agent.
+    
+    Based on A2A Protocol Specification Section 3.1.1.
+    """
+    message: Message
+    configuration: Optional[MessageConfig] = None
+    metadata: Optional[Dict[str, Any]] = None
+
+
+# =============================================================================
+# Streaming Event Types
+# =============================================================================
+
+class TaskStatusUpdateEvent(BaseModel):
+    """
+    Event sent by agent to notify client of task status change.
+    
+    Based on A2A Protocol Specification Section 4.2.1.
+    """
+    model_config = ConfigDict(populate_by_name=True)
+    
+    task_id: str = Field(alias="taskId")
+    context_id: Optional[str] = Field(default=None, alias="contextId")
+    status: TaskStatus
+    final: bool = False
+    metadata: Optional[Dict[str, Any]] = None
+
+
+class TaskArtifactUpdateEvent(BaseModel):
+    """
+    Event sent when an artifact is generated.
+    
+    Based on A2A Protocol Specification Section 4.2.2.
+    """
+    model_config = ConfigDict(populate_by_name=True)
+    
+    task_id: str = Field(alias="taskId")
+    context_id: Optional[str] = Field(default=None, alias="contextId")
+    artifact: Artifact
+    append: bool = False
+    last_chunk: bool = Field(default=False, alias="lastChunk")
+    metadata: Optional[Dict[str, Any]] = None

src/praisonai-agents/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "praisonaiagents"
-version = "0.1.15"
+version = "0.1.16"
 description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
 readme = "README.md"
 requires-python = ">=3.10"

src/praisonai-agents/tests/a2a/__init__.py

@@ -0,0 +1 @@
+"""A2A Protocol Tests"""

src/praisonai-agents/tests/a2a/test_a2a.py

@@ -0,0 +1,135 @@
+"""
+Tests for Main A2A Class
+
+TDD: Write tests first, then implement A2A class.
+"""
+
+
+class TestA2AClass:
+    """Tests for A2A main class."""
+    
+    def test_a2a_creation_with_agent(self):
+        """Test creating A2A with single agent."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a import A2A
+        
+        agent = Agent(name="Test", role="Tester", goal="Test")
+        a2a = A2A(agent=agent)
+        
+        assert a2a.agent is not None
+        assert a2a.name == "Test"
+    
+    def test_a2a_creation_with_custom_name(self):
+        """Test creating A2A with custom name."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a import A2A
+        
+        agent = Agent(name="Test", role="Tester", goal="Test")
+        a2a = A2A(agent=agent, name="Custom Name")
+        
+        assert a2a.name == "Custom Name"
+    
+    def test_a2a_requires_agent(self):
+        """Test A2A requires agent or agents."""
+        from praisonaiagents.ui.a2a import A2A
+        import pytest
+        
+        with pytest.raises(ValueError):
+            A2A()
+    
+    def test_a2a_get_agent_card(self):
+        """Test getting Agent Card from A2A."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a import A2A
+        
+        agent = Agent(name="Card Test", role="Tester", goal="Test")
+        a2a = A2A(agent=agent, url="http://localhost:8000/a2a")
+        
+        card = a2a.get_agent_card()
+        
+        assert card.name == "Card Test"
+        assert card.url == "http://localhost:8000/a2a"
+    
+    def test_a2a_get_router(self):
+        """Test getting FastAPI router from A2A."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a import A2A
+        
+        agent = Agent(name="Router Test", role="Tester", goal="Test")
+        a2a = A2A(agent=agent)
+        
+        router = a2a.get_router()
+        
+        assert router is not None
+        # Check router has routes
+        assert len(router.routes) > 0
+
+
+class TestA2ARouter:
+    """Tests for A2A FastAPI router endpoints."""
+    
+    def test_agent_card_endpoint(self):
+        """Test /.well-known/agent.json endpoint."""
+        from fastapi.testclient import TestClient
+        from fastapi import FastAPI
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a import A2A
+        
+        agent = Agent(name="Endpoint Test", role="Tester", goal="Test")
+        a2a = A2A(agent=agent, url="http://localhost:8000/a2a")
+        
+        app = FastAPI()
+        app.include_router(a2a.get_router())
+        
+        client = TestClient(app)
+        response = client.get("/.well-known/agent.json")
+        
+        assert response.status_code == 200
+        data = response.json()
+        assert data["name"] == "Endpoint Test"
+    
+    def test_status_endpoint(self):
+        """Test /status endpoint."""
+        from fastapi.testclient import TestClient
+        from fastapi import FastAPI
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a import A2A
+        
+        agent = Agent(name="Status Test", role="Tester", goal="Test")
+        a2a = A2A(agent=agent)
+        
+        app = FastAPI()
+        app.include_router(a2a.get_router())
+        
+        client = TestClient(app)
+        response = client.get("/status")
+        
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] == "ok"
+
+
+class TestA2AWithTools:
+    """Tests for A2A with agent tools."""
+    
+    def test_a2a_agent_card_includes_skills(self):
+        """Test Agent Card includes skills from tools."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a import A2A
+        
+        def my_tool(x: str) -> str:
+            """A test tool."""
+            return x
+        
+        agent = Agent(
+            name="Tool Agent",
+            role="Helper",
+            goal="Help",
+            tools=[my_tool]
+        )
+        a2a = A2A(agent=agent, url="http://localhost:8000/a2a")
+        
+        card = a2a.get_agent_card()
+        
+        assert card.skills is not None
+        assert len(card.skills) >= 1

src/praisonai-agents/tests/a2a/test_agent_card.py

@@ -0,0 +1,200 @@
+"""
+Tests for A2A Agent Card Generation
+
+TDD: Write tests first, then implement agent_card module.
+"""
+
+import json
+
+
+class TestAgentCardGeneration:
+    """Tests for generating Agent Card from PraisonAI Agent."""
+    
+    def test_generate_agent_card_basic(self):
+        """Test generating Agent Card from basic Agent."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a.agent_card import generate_agent_card
+        
+        agent = Agent(
+            name="Test Agent",
+            role="Test Role",
+            goal="Test Goal"
+        )
+        
+        card = generate_agent_card(agent, url="http://localhost:8000/a2a")
+        
+        assert card.name == "Test Agent"
+        assert card.url == "http://localhost:8000/a2a"
+        assert card.version == "1.0.0"
+    
+    def test_generate_agent_card_with_description(self):
+        """Test Agent Card includes description from role."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a.agent_card import generate_agent_card
+        
+        agent = Agent(
+            name="Research Agent",
+            role="Research Analyst",
+            goal="Research topics thoroughly"
+        )
+        
+        card = generate_agent_card(agent, url="http://localhost:8000/a2a")
+        
+        assert card.description is not None
+        assert "Research" in card.description
+    
+    def test_generate_agent_card_with_instructions(self):
+        """Test Agent Card from Agent with instructions."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a.agent_card import generate_agent_card
+        
+        agent = Agent(
+            instructions="You are a helpful assistant that answers questions."
+        )
+        
+        card = generate_agent_card(agent, url="http://localhost:8000/a2a")
+        
+        assert card.name is not None
+        assert card.description is not None
+    
+    def test_generate_agent_card_capabilities(self):
+        """Test Agent Card has correct capabilities."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a.agent_card import generate_agent_card
+        
+        agent = Agent(name="Test", role="Tester", goal="Test")
+        
+        card = generate_agent_card(
+            agent, 
+            url="http://localhost:8000/a2a",
+            streaming=True
+        )
+        
+        assert card.capabilities.streaming is True
+
+
+class TestSkillsExtraction:
+    """Tests for extracting skills from Agent tools."""
+    
+    def test_extract_skills_from_function(self):
+        """Test extracting skill from a function tool."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a.agent_card import generate_agent_card
+        
+        def search_web(query: str) -> str:
+            """Search the web for information."""
+            return f"Results for: {query}"
+        
+        agent = Agent(
+            name="Search Agent",
+            role="Searcher",
+            goal="Search the web",
+            tools=[search_web]
+        )
+        
+        card = generate_agent_card(agent, url="http://localhost:8000/a2a")
+        
+        assert card.skills is not None
+        assert len(card.skills) >= 1
+        
+        skill_names = [s.name for s in card.skills]
+        assert "search_web" in skill_names
+    
+    def test_extract_skills_with_description(self):
+        """Test skill description comes from function docstring."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a.agent_card import generate_agent_card
+        
+        def calculate(expression: str) -> str:
+            """Calculate a mathematical expression."""
+            return str(eval(expression))
+        
+        agent = Agent(
+            name="Calculator",
+            role="Math Helper",
+            goal="Do math",
+            tools=[calculate]
+        )
+        
+        card = generate_agent_card(agent, url="http://localhost:8000/a2a")
+        
+        calc_skill = next((s for s in card.skills if s.name == "calculate"), None)
+        assert calc_skill is not None
+        assert "mathematical" in calc_skill.description.lower()
+    
+    def test_extract_skills_multiple_tools(self):
+        """Test extracting skills from multiple tools."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a.agent_card import generate_agent_card
+        
+        def tool_one(x: str) -> str:
+            """First tool."""
+            return x
+        
+        def tool_two(y: str) -> str:
+            """Second tool."""
+            return y
+        
+        agent = Agent(
+            name="Multi Tool Agent",
+            role="Helper",
+            goal="Help",
+            tools=[tool_one, tool_two]
+        )
+        
+        card = generate_agent_card(agent, url="http://localhost:8000/a2a")
+        
+        assert len(card.skills) >= 2
+    
+    def test_no_skills_when_no_tools(self):
+        """Test Agent Card with no tools has empty skills."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a.agent_card import generate_agent_card
+        
+        agent = Agent(
+            name="Simple Agent",
+            role="Helper",
+            goal="Help"
+        )
+        
+        card = generate_agent_card(agent, url="http://localhost:8000/a2a")
+        
+        # Skills should be None or empty list
+        assert card.skills is None or len(card.skills) == 0
+
+
+class TestAgentCardSerialization:
+    """Tests for Agent Card JSON serialization."""
+    
+    def test_agent_card_to_json(self):
+        """Test Agent Card can be serialized to JSON."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a.agent_card import generate_agent_card
+        
+        agent = Agent(name="Test", role="Tester", goal="Test")
+        card = generate_agent_card(agent, url="http://localhost:8000/a2a")
+        
+        json_str = card.model_dump_json()
+        data = json.loads(json_str)
+        
+        assert "name" in data
+        assert "url" in data
+        assert "capabilities" in data
+    
+    def test_agent_card_camel_case_keys(self):
+        """Test Agent Card JSON uses camelCase keys."""
+        from praisonaiagents import Agent
+        from praisonaiagents.ui.a2a.agent_card import generate_agent_card
+        
+        agent = Agent(name="Test", role="Tester", goal="Test")
+        card = generate_agent_card(
+            agent, 
+            url="http://localhost:8000/a2a",
+            streaming=True
+        )
+        
+        # Use by_alias to get camelCase
+        data = card.model_dump(by_alias=True)
+        
+        # Check capabilities uses camelCase
+        assert "pushNotifications" in data["capabilities"]

src/praisonai-agents/tests/a2a/test_conversion.py

@@ -0,0 +1,190 @@
+"""
+Tests for A2A Message Conversion
+
+TDD: Write tests first, then implement conversion module.
+"""
+
+
+class TestA2AToPraisonAI:
+    """Tests for converting A2A messages to PraisonAI format."""
+    
+    def test_convert_user_message(self):
+        """Test converting A2A user message to PraisonAI format."""
+        from praisonaiagents.ui.a2a.conversion import a2a_to_praisonai_messages
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        
+        a2a_msg = Message(
+            message_id="msg-1",
+            role=Role.USER,
+            parts=[TextPart(text="Hello, how are you?")]
+        )
+        
+        result = a2a_to_praisonai_messages([a2a_msg])
+        
+        assert len(result) == 1
+        assert result[0]["role"] == "user"
+        assert result[0]["content"] == "Hello, how are you?"
+    
+    def test_convert_agent_message(self):
+        """Test converting A2A agent message to PraisonAI format."""
+        from praisonaiagents.ui.a2a.conversion import a2a_to_praisonai_messages
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        
+        a2a_msg = Message(
+            message_id="msg-2",
+            role=Role.AGENT,
+            parts=[TextPart(text="I'm doing well, thanks!")]
+        )
+        
+        result = a2a_to_praisonai_messages([a2a_msg])
+        
+        assert len(result) == 1
+        assert result[0]["role"] == "assistant"
+        assert result[0]["content"] == "I'm doing well, thanks!"
+    
+    def test_convert_multiple_parts(self):
+        """Test converting message with multiple text parts."""
+        from praisonaiagents.ui.a2a.conversion import a2a_to_praisonai_messages
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        
+        a2a_msg = Message(
+            message_id="msg-3",
+            role=Role.USER,
+            parts=[
+                TextPart(text="First part."),
+                TextPart(text="Second part.")
+            ]
+        )
+        
+        result = a2a_to_praisonai_messages([a2a_msg])
+        
+        assert "First part" in result[0]["content"]
+        assert "Second part" in result[0]["content"]
+    
+    def test_convert_conversation_history(self):
+        """Test converting full conversation history."""
+        from praisonaiagents.ui.a2a.conversion import a2a_to_praisonai_messages
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        
+        messages = [
+            Message(message_id="m1", role=Role.USER, parts=[TextPart(text="Hi")]),
+            Message(message_id="m2", role=Role.AGENT, parts=[TextPart(text="Hello!")]),
+            Message(message_id="m3", role=Role.USER, parts=[TextPart(text="How are you?")]),
+        ]
+        
+        result = a2a_to_praisonai_messages(messages)
+        
+        assert len(result) == 3
+        assert result[0]["role"] == "user"
+        assert result[1]["role"] == "assistant"
+        assert result[2]["role"] == "user"
+
+
+class TestPraisonAIToA2A:
+    """Tests for converting PraisonAI responses to A2A format."""
+    
+    def test_convert_string_response(self):
+        """Test converting string response to A2A Message."""
+        from praisonaiagents.ui.a2a.conversion import praisonai_to_a2a_message
+        
+        response = "This is the agent's response."
+        
+        result = praisonai_to_a2a_message(response)
+        
+        assert result.role.value == "agent"
+        assert len(result.parts) == 1
+        assert result.parts[0].text == response
+    
+    def test_convert_response_with_context(self):
+        """Test converting response with context_id."""
+        from praisonaiagents.ui.a2a.conversion import praisonai_to_a2a_message
+        
+        response = "Response text"
+        
+        result = praisonai_to_a2a_message(response, context_id="ctx-123")
+        
+        assert result.context_id == "ctx-123"
+    
+    def test_convert_response_with_task(self):
+        """Test converting response with task_id."""
+        from praisonaiagents.ui.a2a.conversion import praisonai_to_a2a_message
+        
+        response = "Response text"
+        
+        result = praisonai_to_a2a_message(response, task_id="task-456")
+        
+        assert result.task_id == "task-456"
+
+
+class TestArtifactConversion:
+    """Tests for converting responses to A2A Artifacts."""
+    
+    def test_create_artifact_from_response(self):
+        """Test creating Artifact from response."""
+        from praisonaiagents.ui.a2a.conversion import create_artifact
+        
+        response = "Generated content here."
+        
+        artifact = create_artifact(response)
+        
+        assert artifact.artifact_id is not None
+        assert len(artifact.parts) == 1
+        assert artifact.parts[0].text == response
+    
+    def test_create_artifact_with_name(self):
+        """Test creating Artifact with name."""
+        from praisonaiagents.ui.a2a.conversion import create_artifact
+        
+        response = "Report content"
+        
+        artifact = create_artifact(response, name="Report", description="Generated report")
+        
+        assert artifact.name == "Report"
+        assert artifact.description == "Generated report"
+
+
+class TestExtractUserInput:
+    """Tests for extracting user input from A2A messages."""
+    
+    def test_extract_from_single_message(self):
+        """Test extracting user input from single message."""
+        from praisonaiagents.ui.a2a.conversion import extract_user_input
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        
+        msg = Message(
+            message_id="m1",
+            role=Role.USER,
+            parts=[TextPart(text="What is AI?")]
+        )
+        
+        result = extract_user_input([msg])
+        
+        assert result == "What is AI?"
+    
+    def test_extract_from_conversation(self):
+        """Test extracting last user input from conversation."""
+        from praisonaiagents.ui.a2a.conversion import extract_user_input
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        
+        messages = [
+            Message(message_id="m1", role=Role.USER, parts=[TextPart(text="First question")]),
+            Message(message_id="m2", role=Role.AGENT, parts=[TextPart(text="Answer")]),
+            Message(message_id="m3", role=Role.USER, parts=[TextPart(text="Follow up question")]),
+        ]
+        
+        result = extract_user_input(messages)
+        
+        assert result == "Follow up question"
+    
+    def test_extract_empty_when_no_user_message(self):
+        """Test extracting returns empty when no user message."""
+        from praisonaiagents.ui.a2a.conversion import extract_user_input
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        
+        messages = [
+            Message(message_id="m1", role=Role.AGENT, parts=[TextPart(text="Hello")]),
+        ]
+        
+        result = extract_user_input(messages)
+        
+        assert result == ""

src/praisonai-agents/tests/a2a/test_streaming.py

@@ -0,0 +1,116 @@
+"""
+Tests for A2A Streaming Events
+"""
+
+
+class TestA2AEventEncoder:
+    """Tests for A2AEventEncoder."""
+    
+    def test_encode_event(self):
+        """Test encoding a generic event."""
+        from praisonaiagents.ui.a2a.streaming import A2AEventEncoder
+        
+        result = A2AEventEncoder.encode_event("test", {"key": "value"})
+        
+        assert "event: test" in result
+        assert "data:" in result
+        assert "key" in result
+    
+    def test_encode_task_status(self):
+        """Test encoding TaskStatusUpdateEvent."""
+        from praisonaiagents.ui.a2a.streaming import A2AEventEncoder
+        from praisonaiagents.ui.a2a.types import TaskStatusUpdateEvent, TaskStatus, TaskState
+        
+        event = TaskStatusUpdateEvent(
+            task_id="task-123",
+            status=TaskStatus(state=TaskState.WORKING),
+            final=False,
+        )
+        
+        result = A2AEventEncoder.encode_task_status(event)
+        
+        assert "event: task.status" in result
+        assert "task-123" in result
+    
+    def test_encode_task_artifact(self):
+        """Test encoding TaskArtifactUpdateEvent."""
+        from praisonaiagents.ui.a2a.streaming import A2AEventEncoder
+        from praisonaiagents.ui.a2a.types import TaskArtifactUpdateEvent, Artifact, TextPart
+        
+        artifact = Artifact(
+            artifact_id="art-1",
+            parts=[TextPart(text="Result")]
+        )
+        event = TaskArtifactUpdateEvent(
+            task_id="task-456",
+            artifact=artifact,
+            last_chunk=True,
+        )
+        
+        result = A2AEventEncoder.encode_task_artifact(event)
+        
+        assert "event: task.artifact" in result
+        assert "art-1" in result
+    
+    def test_encode_done(self):
+        """Test encoding done event."""
+        from praisonaiagents.ui.a2a.streaming import A2AEventEncoder
+        
+        result = A2AEventEncoder.encode_done()
+        
+        assert "event: done" in result
+
+
+class TestCreateEvents:
+    """Tests for event creation helpers."""
+    
+    def test_create_status_event(self):
+        """Test creating status event."""
+        from praisonaiagents.ui.a2a.streaming import create_status_event
+        from praisonaiagents.ui.a2a.types import TaskState
+        
+        event = create_status_event(
+            task_id="task-1",
+            state=TaskState.WORKING,
+        )
+        
+        assert event.task_id == "task-1"
+        assert event.status.state == TaskState.WORKING
+    
+    def test_create_status_event_final(self):
+        """Test creating final status event."""
+        from praisonaiagents.ui.a2a.streaming import create_status_event
+        from praisonaiagents.ui.a2a.types import TaskState
+        
+        event = create_status_event(
+            task_id="task-2",
+            state=TaskState.COMPLETED,
+            final=True,
+        )
+        
+        assert event.final is True
+    
+    def test_create_artifact_event(self):
+        """Test creating artifact event."""
+        from praisonaiagents.ui.a2a.streaming import create_artifact_event
+        
+        event = create_artifact_event(
+            task_id="task-3",
+            content="Generated content",
+        )
+        
+        assert event.task_id == "task-3"
+        assert event.artifact is not None
+        assert len(event.artifact.parts) == 1
+    
+    def test_create_artifact_event_last_chunk(self):
+        """Test creating last chunk artifact event."""
+        from praisonaiagents.ui.a2a.streaming import create_artifact_event
+        
+        event = create_artifact_event(
+            task_id="task-4",
+            content="Final content",
+            last_chunk=True,
+        )
+        
+        assert event.last_chunk is True

src/praisonai-agents/tests/a2a/test_task_store.py

@@ -0,0 +1,225 @@
+"""
+Tests for A2A Task Store
+
+TDD: Write tests first, then implement task_store module.
+"""
+
+
+class TestTaskStore:
+    """Tests for TaskStore class."""
+    
+    def test_create_task(self):
+        """Test creating a new task."""
+        from praisonaiagents.ui.a2a.task_store import TaskStore
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart, TaskState
+        
+        store = TaskStore()
+        msg = Message(
+            message_id="msg-1",
+            role=Role.USER,
+            parts=[TextPart(text="Hello")]
+        )
+        
+        task = store.create_task(msg)
+        
+        assert task.id is not None
+        assert task.status.state == TaskState.SUBMITTED
+    
+    def test_create_task_with_context(self):
+        """Test creating task with context_id."""
+        from praisonaiagents.ui.a2a.task_store import TaskStore
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        
+        store = TaskStore()
+        msg = Message(
+            message_id="msg-2",
+            role=Role.USER,
+            parts=[TextPart(text="Hello")],
+            context_id="ctx-123"
+        )
+        
+        task = store.create_task(msg)
+        
+        assert task.context_id == "ctx-123"
+    
+    def test_get_task(self):
+        """Test retrieving a task by ID."""
+        from praisonaiagents.ui.a2a.task_store import TaskStore
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        
+        store = TaskStore()
+        msg = Message(
+            message_id="msg-3",
+            role=Role.USER,
+            parts=[TextPart(text="Hello")]
+        )
+        
+        created = store.create_task(msg)
+        retrieved = store.get_task(created.id)
+        
+        assert retrieved is not None
+        assert retrieved.id == created.id
+    
+    def test_get_task_not_found(self):
+        """Test getting non-existent task returns None."""
+        from praisonaiagents.ui.a2a.task_store import TaskStore
+        
+        store = TaskStore()
+        result = store.get_task("non-existent-id")
+        
+        assert result is None
+    
+    def test_update_status(self):
+        """Test updating task status."""
+        from praisonaiagents.ui.a2a.task_store import TaskStore
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart, TaskState
+        
+        store = TaskStore()
+        msg = Message(
+            message_id="msg-4",
+            role=Role.USER,
+            parts=[TextPart(text="Hello")]
+        )
+        
+        task = store.create_task(msg)
+        updated = store.update_status(task.id, TaskState.WORKING)
+        
+        assert updated.status.state == TaskState.WORKING
+    
+    def test_update_status_to_completed(self):
+        """Test updating task to completed state."""
+        from praisonaiagents.ui.a2a.task_store import TaskStore
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart, TaskState
+        
+        store = TaskStore()
+        msg = Message(
+            message_id="msg-5",
+            role=Role.USER,
+            parts=[TextPart(text="Hello")]
+        )
+        
+        task = store.create_task(msg)
+        store.update_status(task.id, TaskState.WORKING)
+        updated = store.update_status(task.id, TaskState.COMPLETED)
+        
+        assert updated.status.state == TaskState.COMPLETED
+    
+    def test_add_artifact(self):
+        """Test adding artifact to task."""
+        from praisonaiagents.ui.a2a.task_store import TaskStore
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart, Artifact
+        
+        store = TaskStore()
+        msg = Message(
+            message_id="msg-6",
+            role=Role.USER,
+            parts=[TextPart(text="Hello")]
+        )
+        
+        task = store.create_task(msg)
+        artifact = Artifact(
+            artifact_id="art-1",
+            parts=[TextPart(text="Result")]
+        )
+        
+        updated = store.add_artifact(task.id, artifact)
+        
+        assert updated.artifacts is not None
+        assert len(updated.artifacts) == 1
+    
+    def test_add_message_to_history(self):
+        """Test adding message to task history."""
+        from praisonaiagents.ui.a2a.task_store import TaskStore
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        
+        store = TaskStore()
+        msg1 = Message(
+            message_id="msg-7",
+            role=Role.USER,
+            parts=[TextPart(text="Question")]
+        )
+        
+        task = store.create_task(msg1)
+        
+        msg2 = Message(
+            message_id="msg-8",
+            role=Role.AGENT,
+            parts=[TextPart(text="Answer")]
+        )
+        
+        updated = store.add_to_history(task.id, msg2)
+        
+        assert updated.history is not None
+        assert len(updated.history) >= 1
+    
+    def test_list_tasks(self):
+        """Test listing all tasks."""
+        from praisonaiagents.ui.a2a.task_store import TaskStore
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        
+        store = TaskStore()
+        
+        for i in range(3):
+            msg = Message(
+                message_id=f"msg-{i}",
+                role=Role.USER,
+                parts=[TextPart(text=f"Message {i}")]
+            )
+            store.create_task(msg)
+        
+        tasks = store.list_tasks()
+        
+        assert len(tasks) == 3
+    
+    def test_cancel_task(self):
+        """Test cancelling a task."""
+        from praisonaiagents.ui.a2a.task_store import TaskStore
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart, TaskState
+        
+        store = TaskStore()
+        msg = Message(
+            message_id="msg-cancel",
+            role=Role.USER,
+            parts=[TextPart(text="Hello")]
+        )
+        
+        task = store.create_task(msg)
+        cancelled = store.cancel_task(task.id)
+        
+        assert cancelled.status.state == TaskState.CANCELLED
+
+
+class TestTaskStoreContextManagement:
+    """Tests for context management in TaskStore."""
+    
+    def test_tasks_grouped_by_context(self):
+        """Test getting tasks by context_id."""
+        from praisonaiagents.ui.a2a.task_store import TaskStore
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        
+        store = TaskStore()
+        
+        # Create tasks in context A
+        for i in range(2):
+            msg = Message(
+                message_id=f"msg-a-{i}",
+                role=Role.USER,
+                parts=[TextPart(text=f"A{i}")],
+                context_id="context-a"
+            )
+            store.create_task(msg)
+        
+        # Create task in context B
+        msg_b = Message(
+            message_id="msg-b-1",
+            role=Role.USER,
+            parts=[TextPart(text="B1")],
+            context_id="context-b"
+        )
+        store.create_task(msg_b)
+        
+        tasks_a = store.list_tasks(context_id="context-a")
+        tasks_b = store.list_tasks(context_id="context-b")
+        
+        assert len(tasks_a) == 2
+        assert len(tasks_b) == 1

src/praisonai-agents/tests/a2a/test_types.py

@@ -0,0 +1,395 @@
+"""
+Tests for A2A Protocol Types
+
+TDD: Write tests first, then implement types.
+"""
+
+import json
+from datetime import datetime
+
+
+class TestTaskState:
+    """Tests for TaskState enum."""
+    
+    def test_task_state_submitted(self):
+        """Test SUBMITTED state exists."""
+        from praisonaiagents.ui.a2a.types import TaskState
+        assert TaskState.SUBMITTED == "submitted"
+    
+    def test_task_state_working(self):
+        """Test WORKING state exists."""
+        from praisonaiagents.ui.a2a.types import TaskState
+        assert TaskState.WORKING == "working"
+    
+    def test_task_state_completed(self):
+        """Test COMPLETED state exists."""
+        from praisonaiagents.ui.a2a.types import TaskState
+        assert TaskState.COMPLETED == "completed"
+    
+    def test_task_state_failed(self):
+        """Test FAILED state exists."""
+        from praisonaiagents.ui.a2a.types import TaskState
+        assert TaskState.FAILED == "failed"
+    
+    def test_task_state_cancelled(self):
+        """Test CANCELLED state exists."""
+        from praisonaiagents.ui.a2a.types import TaskState
+        assert TaskState.CANCELLED == "cancelled"
+    
+    def test_task_state_input_required(self):
+        """Test INPUT_REQUIRED state exists."""
+        from praisonaiagents.ui.a2a.types import TaskState
+        assert TaskState.INPUT_REQUIRED == "input_required"
+    
+    def test_task_state_auth_required(self):
+        """Test AUTH_REQUIRED state exists."""
+        from praisonaiagents.ui.a2a.types import TaskState
+        assert TaskState.AUTH_REQUIRED == "auth_required"
+
+
+class TestRole:
+    """Tests for Role enum."""
+    
+    def test_role_user(self):
+        """Test USER role exists."""
+        from praisonaiagents.ui.a2a.types import Role
+        assert Role.USER == "user"
+    
+    def test_role_agent(self):
+        """Test AGENT role exists."""
+        from praisonaiagents.ui.a2a.types import Role
+        assert Role.AGENT == "agent"
+
+
+class TestPart:
+    """Tests for Part models."""
+    
+    def test_text_part_creation(self):
+        """Test TextPart can be created."""
+        from praisonaiagents.ui.a2a.types import TextPart
+        part = TextPart(text="Hello, world!")
+        assert part.text == "Hello, world!"
+    
+    def test_file_part_with_uri(self):
+        """Test FilePart with URI."""
+        from praisonaiagents.ui.a2a.types import FilePart
+        part = FilePart(
+            file_uri="https://example.com/file.pdf",
+            media_type="application/pdf",
+            name="document.pdf"
+        )
+        assert part.file_uri == "https://example.com/file.pdf"
+        assert part.media_type == "application/pdf"
+    
+    def test_file_part_with_bytes(self):
+        """Test FilePart with bytes."""
+        from praisonaiagents.ui.a2a.types import FilePart
+        part = FilePart(
+            file_bytes=b"file content",
+            media_type="text/plain",
+            name="test.txt"
+        )
+        assert part.file_bytes == b"file content"
+    
+    def test_data_part_creation(self):
+        """Test DataPart can be created."""
+        from praisonaiagents.ui.a2a.types import DataPart
+        part = DataPart(data={"key": "value", "count": 42})
+        assert part.data == {"key": "value", "count": 42}
+
+
+class TestMessage:
+    """Tests for Message model."""
+    
+    def test_message_creation(self):
+        """Test Message can be created."""
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        msg = Message(
+            message_id="msg-123",
+            role=Role.USER,
+            parts=[TextPart(text="Hello")]
+        )
+        assert msg.message_id == "msg-123"
+        assert msg.role == Role.USER
+        assert len(msg.parts) == 1
+    
+    def test_message_with_context(self):
+        """Test Message with context_id."""
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        msg = Message(
+            message_id="msg-456",
+            role=Role.AGENT,
+            parts=[TextPart(text="Response")],
+            context_id="ctx-789"
+        )
+        assert msg.context_id == "ctx-789"
+    
+    def test_message_with_task(self):
+        """Test Message with task_id."""
+        from praisonaiagents.ui.a2a.types import Message, Role, TextPart
+        msg = Message(
+            message_id="msg-789",
+            role=Role.AGENT,
+            parts=[TextPart(text="Working on it")],
+            task_id="task-123"
+        )
+        assert msg.task_id == "task-123"
+
+
+class TestTaskStatus:
+    """Tests for TaskStatus model."""
+    
+    def test_task_status_creation(self):
+        """Test TaskStatus can be created."""
+        from praisonaiagents.ui.a2a.types import TaskStatus, TaskState
+        status = TaskStatus(state=TaskState.WORKING)
+        assert status.state == TaskState.WORKING
+    
+    def test_task_status_with_message(self):
+        """Test TaskStatus with message."""
+        from praisonaiagents.ui.a2a.types import TaskStatus, TaskState, Message, Role, TextPart
+        msg = Message(
+            message_id="msg-1",
+            role=Role.AGENT,
+            parts=[TextPart(text="Processing...")]
+        )
+        status = TaskStatus(state=TaskState.WORKING, message=msg)
+        assert status.message is not None
+    
+    def test_task_status_with_timestamp(self):
+        """Test TaskStatus with timestamp."""
+        from praisonaiagents.ui.a2a.types import TaskStatus, TaskState
+        ts = datetime.utcnow().isoformat()
+        status = TaskStatus(state=TaskState.COMPLETED, timestamp=ts)
+        assert status.timestamp == ts
+
+
+class TestTask:
+    """Tests for Task model."""
+    
+    def test_task_creation(self):
+        """Test Task can be created."""
+        from praisonaiagents.ui.a2a.types import Task, TaskStatus, TaskState
+        task = Task(
+            id="task-123",
+            status=TaskStatus(state=TaskState.SUBMITTED)
+        )
+        assert task.id == "task-123"
+        assert task.status.state == TaskState.SUBMITTED
+    
+    def test_task_with_context(self):
+        """Test Task with context_id."""
+        from praisonaiagents.ui.a2a.types import Task, TaskStatus, TaskState
+        task = Task(
+            id="task-456",
+            context_id="ctx-789",
+            status=TaskStatus(state=TaskState.WORKING)
+        )
+        assert task.context_id == "ctx-789"
+    
+    def test_task_with_artifacts(self):
+        """Test Task with artifacts."""
+        from praisonaiagents.ui.a2a.types import Task, TaskStatus, TaskState, Artifact, TextPart
+        artifact = Artifact(
+            artifact_id="art-1",
+            parts=[TextPart(text="Result")]
+        )
+        task = Task(
+            id="task-789",
+            status=TaskStatus(state=TaskState.COMPLETED),
+            artifacts=[artifact]
+        )
+        assert len(task.artifacts) == 1
+    
+    def test_task_with_history(self):
+        """Test Task with message history."""
+        from praisonaiagents.ui.a2a.types import Task, TaskStatus, TaskState, Message, Role, TextPart
+        msg1 = Message(message_id="m1", role=Role.USER, parts=[TextPart(text="Q")])
+        msg2 = Message(message_id="m2", role=Role.AGENT, parts=[TextPart(text="A")])
+        task = Task(
+            id="task-abc",
+            status=TaskStatus(state=TaskState.COMPLETED),
+            history=[msg1, msg2]
+        )
+        assert len(task.history) == 2
+
+
+class TestArtifact:
+    """Tests for Artifact model."""
+    
+    def test_artifact_creation(self):
+        """Test Artifact can be created."""
+        from praisonaiagents.ui.a2a.types import Artifact, TextPart
+        artifact = Artifact(
+            artifact_id="art-123",
+            parts=[TextPart(text="Generated content")]
+        )
+        assert artifact.artifact_id == "art-123"
+    
+    def test_artifact_with_name(self):
+        """Test Artifact with name and description."""
+        from praisonaiagents.ui.a2a.types import Artifact, TextPart
+        artifact = Artifact(
+            artifact_id="art-456",
+            name="Report",
+            description="Generated report",
+            parts=[TextPart(text="Report content")]
+        )
+        assert artifact.name == "Report"
+        assert artifact.description == "Generated report"
+
+
+class TestAgentSkill:
+    """Tests for AgentSkill model."""
+    
+    def test_agent_skill_creation(self):
+        """Test AgentSkill can be created."""
+        from praisonaiagents.ui.a2a.types import AgentSkill
+        skill = AgentSkill(
+            id="search",
+            name="Web Search",
+            description="Search the web for information"
+        )
+        assert skill.id == "search"
+        assert skill.name == "Web Search"
+    
+    def test_agent_skill_with_tags(self):
+        """Test AgentSkill with tags."""
+        from praisonaiagents.ui.a2a.types import AgentSkill
+        skill = AgentSkill(
+            id="calc",
+            name="Calculator",
+            description="Perform calculations",
+            tags=["math", "utility"]
+        )
+        assert skill.tags == ["math", "utility"]
+    
+    def test_agent_skill_with_examples(self):
+        """Test AgentSkill with examples."""
+        from praisonaiagents.ui.a2a.types import AgentSkill
+        skill = AgentSkill(
+            id="translate",
+            name="Translator",
+            description="Translate text",
+            examples=["Translate hello to Spanish", "What is bonjour in English?"]
+        )
+        assert len(skill.examples) == 2
+
+
+class TestAgentCapabilities:
+    """Tests for AgentCapabilities model."""
+    
+    def test_capabilities_creation(self):
+        """Test AgentCapabilities can be created."""
+        from praisonaiagents.ui.a2a.types import AgentCapabilities
+        caps = AgentCapabilities(streaming=True)
+        assert caps.streaming is True
+    
+    def test_capabilities_defaults(self):
+        """Test AgentCapabilities defaults."""
+        from praisonaiagents.ui.a2a.types import AgentCapabilities
+        caps = AgentCapabilities()
+        assert caps.streaming is False
+        assert caps.push_notifications is False
+
+
+class TestAgentCard:
+    """Tests for AgentCard model."""
+    
+    def test_agent_card_creation(self):
+        """Test AgentCard can be created."""
+        from praisonaiagents.ui.a2a.types import AgentCard, AgentCapabilities
+        card = AgentCard(
+            name="Research Agent",
+            url="http://localhost:8000/a2a",
+            version="1.0.0",
+            capabilities=AgentCapabilities(streaming=True)
+        )
+        assert card.name == "Research Agent"
+        assert card.url == "http://localhost:8000/a2a"
+    
+    def test_agent_card_with_skills(self):
+        """Test AgentCard with skills."""
+        from praisonaiagents.ui.a2a.types import AgentCard, AgentCapabilities, AgentSkill
+        skill = AgentSkill(id="search", name="Search", description="Search web")
+        card = AgentCard(
+            name="Search Agent",
+            url="http://localhost:8000/a2a",
+            version="1.0.0",
+            capabilities=AgentCapabilities(),
+            skills=[skill]
+        )
+        assert len(card.skills) == 1
+    
+    def test_agent_card_json_serialization(self):
+        """Test AgentCard can be serialized to JSON."""
+        from praisonaiagents.ui.a2a.types import AgentCard, AgentCapabilities
+        card = AgentCard(
+            name="Test Agent",
+            url="http://localhost:8000/a2a",
+            version="1.0.0",
+            capabilities=AgentCapabilities()
+        )
+        json_str = card.model_dump_json()
+        data = json.loads(json_str)
+        assert data["name"] == "Test Agent"
+
+
+class TestSendMessageRequest:
+    """Tests for SendMessageRequest model."""
+    
+    def test_send_message_request_creation(self):
+        """Test SendMessageRequest can be created."""
+        from praisonaiagents.ui.a2a.types import SendMessageRequest, Message, Role, TextPart
+        msg = Message(
+            message_id="msg-1",
+            role=Role.USER,
+            parts=[TextPart(text="Hello")]
+        )
+        req = SendMessageRequest(message=msg)
+        assert req.message.message_id == "msg-1"
+    
+    def test_send_message_request_with_config(self):
+        """Test SendMessageRequest with configuration."""
+        from praisonaiagents.ui.a2a.types import SendMessageRequest, Message, Role, TextPart, MessageConfig
+        msg = Message(
+            message_id="msg-2",
+            role=Role.USER,
+            parts=[TextPart(text="Search for AI")]
+        )
+        config = MessageConfig(
+            accept_content_types=["text/plain", "application/json"]
+        )
+        req = SendMessageRequest(message=msg, configuration=config)
+        assert req.configuration is not None
+
+
+class TestStreamingEvents:
+    """Tests for streaming event models."""
+    
+    def test_task_status_update_event(self):
+        """Test TaskStatusUpdateEvent creation."""
+        from praisonaiagents.ui.a2a.types import TaskStatusUpdateEvent, TaskStatus, TaskState
+        event = TaskStatusUpdateEvent(
+            task_id="task-123",
+            status=TaskStatus(state=TaskState.WORKING),
+            final=False
+        )
+        assert event.task_id == "task-123"
+        assert event.final is False
+    
+    def test_task_artifact_update_event(self):
+        """Test TaskArtifactUpdateEvent creation."""
+        from praisonaiagents.ui.a2a.types import TaskArtifactUpdateEvent, Artifact, TextPart
+        artifact = Artifact(
+            artifact_id="art-1",
+            parts=[TextPart(text="Partial result")]
+        )
+        event = TaskArtifactUpdateEvent(
+            task_id="task-456",
+            artifact=artifact,
+            append=True,
+            last_chunk=False
+        )
+        assert event.append is True
+        assert event.last_chunk is False

src/praisonai-agents/uv.lock

@@ -2422,7 +2422,7 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "0.1.15"
+version = "0.1.16"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },

src/praisonai/README.md

@@ -1628,6 +1628,48 @@ server.run()  # stdio for Claude Desktop
 | Security | Origin validation, DNS rebinding prevention |
 | WebSocket | Auto-reconnect with exponential backoff |
 
+## A2A (Agent2Agent Protocol)
+
+PraisonAI supports the [A2A Protocol](https://a2a-protocol.org) for agent-to-agent communication, enabling your agents to be discovered and collaborate with other AI agents.
+
+### A2A Server (Expose Agent as A2A Server)
+```python
+from praisonaiagents import Agent, A2A
+from fastapi import FastAPI
+
+# Create an agent with tools
+def search_web(query: str) -> str:
+    """Search the web for information."""
+    return f"Results for: {query}"
+
+agent = Agent(
+    name="Research Assistant",
+    role="Research Analyst",
+    goal="Help users research topics",
+    tools=[search_web]
+)
+
+# Expose as A2A Server
+a2a = A2A(agent=agent, url="http://localhost:8000/a2a")
+
+app = FastAPI()
+app.include_router(a2a.get_router())
+
+# Run: uvicorn app:app --reload
+# Agent Card: GET /.well-known/agent.json
+# Status: GET /status
+```
+
+### A2A Features
+| Feature | Description |
+|---------|-------------|
+| Agent Card | JSON metadata for agent discovery |
+| Skills Extraction | Auto-generate skills from tools |
+| Task Management | Stateful task lifecycle |
+| Streaming | SSE streaming for real-time updates |
+
+> **Documentation**: [docs.praison.ai/a2a](https://docs.praison.ai/a2a) | **Examples**: [examples/python/a2a](https://github.com/MervinPraison/PraisonAI/tree/main/examples/python/a2a)
+
 ## CLI Features
 
 | Feature | Docs |

src/praisonai/praisonai.rb

@@ -3,8 +3,8 @@ class Praisonai < Formula
   
     desc "AI tools for various AI applications"
     homepage "https://github.com/MervinPraison/PraisonAI"
-    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.37.tar.gz"
-    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.37.tar.gz | shasum -a 256`.split.first
+    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.38.tar.gz"
+    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.38.tar.gz | shasum -a 256`.split.first
     license "MIT"
   
     depends_on "[email protected]"

src/praisonai/praisonai/deploy.py

@@ -57,7 +57,7 @@ class CloudDeployer:
             file.write("FROM python:3.11-slim\n")
             file.write("WORKDIR /app\n")
             file.write("COPY . .\n")
-            file.write("RUN pip install flask praisonai==2.3.37 gunicorn markdown\n")
+            file.write("RUN pip install flask praisonai==2.3.38 gunicorn markdown\n")
             file.write("EXPOSE 8080\n")
             file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
             

src/praisonai/praisonai/version.py

@@ -1 +1 @@
-__version__ = "2.3.37"
+__version__ = "2.3.38"

src/praisonai/pyproject.toml

@@ -12,7 +12,7 @@ dependencies = [
     "rich>=13.7",
     "markdown>=3.5",
     "pyparsing>=3.0.0",
-    "praisonaiagents>=0.1.15",
+    "praisonaiagents>=0.1.16",
     "python-dotenv>=0.19.0",
     "instructor>=1.3.3",
     "PyYAML>=6.0",

src/praisonai/uv.lock

@@ -4197,7 +4197,7 @@ requires-dist = [
     { name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.0.22" },
     { name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.0.22" },
     { name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.0.22" },
-    { name = "praisonaiagents", specifier = ">=0.1.15" },
+    { name = "praisonaiagents", specifier = ">=0.1.16" },
     { name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
     { name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
     { name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -4252,7 +4252,7 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "0.1.15"
+version = "0.1.16"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "aiohttp" },
@@ -4261,9 +4261,9 @@ dependencies = [
     { name = "pydantic" },
     { name = "rich" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/76/1b/9a6352623f2f2c52373a890d04db1d4adc852e164c33840c0071f7d93a50/praisonaiagents-0.1.15.tar.gz", hash = "sha256:45c2ea0a699df4c28a10bf192ef9f6fde1bf9178e966a9892148ed02f0ddb142", size = 628803, upload-time = "2025-12-18T17:38:14.311Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/43/b6/2de2d2af1eb7e26a4e049c5f9e77931b4d2584ccb28de9c4f3d33b386ba2/praisonaiagents-0.1.16.tar.gz", hash = "sha256:bcbe72de8939849de193383e499604e98efdbe8c47d5f47c180ae7cc552d2f26", size = 637000, upload-time = "2025-12-18T21:23:39.359Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/a4/e1/d0a98166a357da2ff94c79f3c311b070a8f37cd7bbec23f7648a29efa417/praisonaiagents-0.1.15-py3-none-any.whl", hash = "sha256:596991445a271e70831a89d300a25f8e482d75ef0405cab22eb5f53293880af1", size = 557414, upload-time = "2025-12-18T17:38:12.296Z" },
+    { url = "https://files.pythonhosted.org/packages/7a/ae/c133a8245d2c2e041fd11cb9a3b2e760ae51caea9ed0bb7a55611494aaad/praisonaiagents-0.1.16-py3-none-any.whl", hash = "sha256:2d98c7bdf7af7ac903198835a4874b3d353d170eae5472d977c0ecaccd9bbe94", size = 567526, upload-time = "2025-12-18T21:23:37.549Z" },
 ]
 
 [[package]]