Initial backend setup with core services and API endpoints

Author: KB01111

app/__init__.py

@@ -0,0 +1,3 @@
+"""Main application package for the Atlas AgentVerse backend."""
+
+

app/api/__init__.py

@@ -0,0 +1,3 @@
+"""API subpackage containing routers and endpoints."""
+
+

app/api/endpoints/__init__.py

@@ -0,0 +1,3 @@
+"""Endpoints subpackage containing specific API route implementations."""
+
+

app/api/endpoints/a2a.py

@@ -1,17 +1,20 @@
-from __future__ import annotations
+"""API Endpoints for Google Agent-to-Agent (A2A) Communication Protocol.
 
+Implements registration, handshake, and message relay according to the A2A spec.
+Ref: https://google.github.io/A2A/#/documentation
 """
-A2A Endpoints: All endpoints in this file implement the Google Agent2Agent (A2A) protocol.
-Any reference to 'a2a' refers exclusively to Google A2A: https://google.github.io/A2A/#/documentation
-"""
 
-from fastapi import APIRouter, Depends, HTTPException, Request
+import logging
+from typing import Annotated, Optional
+
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, status
 
 from app.core.security import get_current_user_id
 from app.models.a2a import (
     A2AErrorResponse,
     A2AHandshakeRequest,
     A2AHandshakeResponse,
+    A2AMessage,
     A2ARegistrationRequest,
     A2ARegistrationResponse,
     A2ARequest,
@@ -20,23 +23,119 @@
 )
 from app.services.a2a_service import A2AService
 
+logger = logging.getLogger(__name__)
+
+
+def get_a2a_service():
+    """FastAPI dependency injector for A2AService."""
+    return A2AService()
+
+
 router = APIRouter(prefix="/api/v1/a2a", tags=["a2a"])
 
+
 @router.post("/register", response_model=A2ARegistrationResponse)
-async def register_agent(payload: A2ARegistrationRequest):
-    """Google A2A: Register this agent with the registry or another agent."""
+async def register_agent(
+    request: A2ARegistrationRequest,
+    a2a_service: Annotated[A2AService, Depends(get_a2a_service)],
+    # current_user: Annotated[User, Depends(get_current_active_user)] # TODO: Add auth
+) -> A2ARegistrationResponse:
+    """Register an agent instance with the A2A service.
+
+    Allows an agent to announce its presence and capabilities.
+    """
+    logger.info(f"Received A2A registration request for agent: {request.agent_id}")
     try:
-        return await A2AService.register_agent(payload)
-    except Exception as exc:
-        raise HTTPException(status_code=500, detail=str(exc)) from exc
+        response = await a2a_service.handle_registration(request)
+        logger.info(f"Agent {request.agent_id} registered successfully.")
+        return response
+    except Exception as e:
+        logger.exception(f"Error during A2A registration for agent {request.agent_id}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Registration failed: {e}",
+        ) from e
+
 
 @router.post("/handshake", response_model=A2AHandshakeResponse)
-async def handshake(payload: A2AHandshakeRequest):
-    """Google A2A: Perform handshake/authentication with another agent."""
+async def perform_handshake(
+    request: A2AHandshakeRequest,
+    a2a_service: Annotated[A2AService, Depends(get_a2a_service)],
+    # current_user: Annotated[User, Depends(get_current_active_user)] # TODO: Add auth
+) -> A2AHandshakeResponse:
+    """Initiate or respond to an A2A handshake between agents.
+
+    Establishes secure communication channels.
+    """
+    logger.info(
+        f"Received A2A handshake request from {request.sender_id} "
+        f"to {request.receiver_id}"
+    )
     try:
-        return await A2AService.handshake(payload)
-    except Exception as exc:
-        raise HTTPException(status_code=500, detail=str(exc)) from exc
+        response = await a2a_service.handle_handshake(request)
+        logger.info(
+            f"A2A handshake successful between {request.sender_id} "
+            f"and {request.receiver_id}"
+        )
+        return response
+    except ValueError as ve:
+        logger.warning(f"A2A handshake validation error: {ve}")
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST, detail=f"Handshake error: {ve}"
+        ) from ve
+    except Exception as e:
+        logger.exception(
+            f"Error during A2A handshake between {request.sender_id} "
+            f"and {request.receiver_id}"
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Handshake failed: {e}",
+        ) from e
+
+
+@router.post("/message", status_code=status.HTTP_202_ACCEPTED)
+async def relay_message(
+    request: A2AMessage,
+    a2a_service: Annotated[A2AService, Depends(get_a2a_service)],
+    background_tasks: BackgroundTasks,
+    # current_user: Annotated[User, Depends(get_current_active_user)] # TODO: Add auth
+) -> dict[str, str]:
+    """Receive and relay an A2A message to the intended recipient agent.
+
+    Uses background tasks for asynchronous delivery.
+    """
+    logger.info(
+        f"Received A2A message relay request from {request.sender_id} "
+        f"to {request.receiver_id}"
+    )
+    try:
+        # Validate message structure before queueing
+        # Minimal validation here; deeper validation in the service
+        if not request.sender_id or not request.receiver_id or not request.payload:
+            raise ValueError("Invalid A2A message structure")
+
+        background_tasks.add_task(a2a_service.handle_message_relay, request)
+        logger.info(
+            f"A2A message from {request.sender_id} to {request.receiver_id} "
+            f"queued for relay."
+        )
+        return {"status": "Message relay accepted"}
+    except ValueError as ve:
+        logger.warning(f"A2A message relay validation error: {ve}")
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST, detail=f"Message relay error: {ve}"
+        ) from ve
+    except Exception as e:
+        logger.exception(
+            f"Error queueing A2A message relay from {request.sender_id} "
+            f"to {request.receiver_id}"
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Message relay failed: {e}",
+        ) from e
+
 
 @router.post("/send", response_model=A2AResponse)
 async def send_a2a(
@@ -51,6 +150,7 @@ async def send_a2a(
     except Exception as exc:
         raise HTTPException(status_code=500, detail=str(exc)) from exc
 
+
 @router.post("/receive", response_model=A2AResponse)
 async def receive_a2a(
     request: Request,
@@ -64,6 +164,7 @@ async def receive_a2a(
     except Exception as exc:
         raise HTTPException(status_code=500, detail=str(exc)) from exc
 
+
 @router.get("/status/{agent_id}", response_model=A2AStatusResponse)
 async def agent_status(agent_id: str):
     """Google A2A: Get the current status of this agent."""
@@ -72,7 +173,11 @@ async def agent_status(agent_id: str):
     except Exception as exc:
         raise HTTPException(status_code=500, detail=str(exc)) from exc
 
-@router.get("/error", response_model=A2AErrorResponse)
-async def error_response(error: str, code: int = None):
+
+@router.post("/error", response_model=A2AErrorResponse)
+async def error_response(error: str, code: Optional[int] = None):
     """Google A2A: Return a protocol-compliant error response."""
-    return await A2AService.error_response(error, code)
+    try:
+        return await A2AService.error_response(error, code)
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=str(exc)) from exc

app/api/endpoints/agents.py

@@ -14,28 +14,23 @@
 
 get_current_user_id = Depends(_get_current_user_id)
 
+
 @router.post("/", response_model=AgentOut, status_code=status.HTTP_201_CREATED)
-async def create_agent_endpoint(
-    agent: AgentCreate,
-    user_id: str = get_current_user_id
-):
+async def create_agent_endpoint(agent: AgentCreate, user_id: str = get_current_user_id):
     """Create a new agent for the current user."""
     db_agent = await create_agent(user_id, agent)
     return db_agent
 
+
 @router.get("/", response_model=list[AgentOut])
-async def list_agents_endpoint(
-    user_id: str = get_current_user_id
-):
+async def list_agents_endpoint(user_id: str = get_current_user_id):
     """List all agents for the current user."""
     agents = await get_agents_by_user(user_id)
     return agents
 
+
 @router.get("/{agent_id}", response_model=AgentOut)
-async def get_agent_endpoint(
-    agent_id: str,
-    user_id: str = get_current_user_id
-):
+async def get_agent_endpoint(agent_id: str, user_id: str = get_current_user_id):
     """Retrieve a specific agent by ID for the current user."""
     agent = await get_agent_by_id(agent_id, user_id)
     if not agent:
@@ -45,11 +40,10 @@ async def get_agent_endpoint(
         )
     return agent
 
+
 @router.put("/{agent_id}", response_model=AgentOut)
 async def update_agent_endpoint(
-    agent_id: str,
-    agent_update: AgentUpdate,
-    user_id: str = get_current_user_id
+    agent_id: str, agent_update: AgentUpdate, user_id: str = get_current_user_id
 ):
     """Update an agent by ID for the current user."""
     updated = await update_agent(agent_id, user_id, agent_update)
@@ -60,11 +54,9 @@ async def update_agent_endpoint(
         )
     return updated
 
+
 @router.delete("/{agent_id}", response_model=AgentOut)
-async def archive_agent_endpoint(
-    agent_id: str,
-    user_id: str = get_current_user_id
-):
+async def archive_agent_endpoint(agent_id: str, user_id: str = get_current_user_id):
     """Archive (delete) an agent by ID for the current user."""
     archived = await archive_agent(agent_id, user_id)
     if not archived:

app/api/endpoints/chat.py

@@ -12,31 +12,31 @@
 
 router = APIRouter(prefix="/chat", tags=["chat"])
 
+
 @router.post(
     "/sessions",
     response_model=ChatSessionOut,
     status_code=status.HTTP_201_CREATED,
 )
 async def create_session_endpoint(
-    session: ChatSessionCreate,
-    user_id: str = Depends(get_current_user_id)
+    session: ChatSessionCreate, user_id: str = Depends(get_current_user_id)
 ):
     db_session = await create_session(user_id, session)
     return db_session
 
+
 @router.get("/sessions", response_model=List[ChatSessionOut])
-async def list_sessions_endpoint(
-    user_id: str = Depends(get_current_user_id)
-):
+async def list_sessions_endpoint(user_id: str = Depends(get_current_user_id)):
     sessions = await list_sessions(user_id)
     return sessions
 
+
 @router.get("/sessions/{session_id}/messages", response_model=List[ChatMessageOut])
 async def get_messages_endpoint(
     session_id: str,
     limit: int = Query(50, ge=1, le=100),
     offset: int = Query(0, ge=0),
-    user_id: str = Depends(get_current_user_id)
+    user_id: str = Depends(get_current_user_id),
 ):
     # (Optional: check session ownership here)
     messages = await get_messages(session_id, limit=limit, offset=offset)

app/api/endpoints/keys.py

@@ -12,34 +12,31 @@
 
 router = APIRouter(prefix="/keys", tags=["keys"])
 
+
 class APIKeyIn(BaseModel):
     service: str
     key: str
 
+
 class APIKeyMetaOut(BaseModel):
     id: str
     service: str
     created_at: Optional[str]
     last_used_at: Optional[str]
 
+
 @router.post("/", status_code=status.HTTP_201_CREATED)
-async def add_key(
-    key_in: APIKeyIn,
-    user_id: str = Depends(get_current_user_id)
-):
+async def add_key(key_in: APIKeyIn, user_id: str = Depends(get_current_user_id)):
     await store_api_key(user_id, key_in.service, key_in.key)
     return {"status": "ok"}
 
+
 @router.get("/", response_model=List[APIKeyMetaOut])
-async def get_keys(
-    user_id: str = Depends(get_current_user_id)
-):
+async def get_keys(user_id: str = Depends(get_current_user_id)):
     return await list_api_keys(user_id)
 
+
 @router.delete("/{key_id}", status_code=status.HTTP_204_NO_CONTENT)
-async def remove_key(
-    key_id: str,
-    user_id: str = Depends(get_current_user_id)
-):
+async def remove_key(key_id: str, user_id: str = Depends(get_current_user_id)):
     await delete_api_key(user_id, key_id)
     return