✨ Supports northbound apis service

Author: Phinease

backend/apps/agent_app.py

@@ -46,7 +46,7 @@ async def search_agent_info_api(agent_id: int = Body(...), authorization: Option
     """
     try:
         _, tenant_id = get_current_user_id(authorization)
-        return get_agent_info_impl(agent_id, tenant_id)
+        return await get_agent_info_impl(agent_id, tenant_id)
     except Exception as e:
         raise HTTPException(status_code=500, detail=f"Agent search info error: {str(e)}")
 
@@ -57,7 +57,7 @@ async def get_creating_sub_agent_info_api(authorization: Optional[str] = Header(
     Create a new sub agent, return agent_ID
     """
     try:
-        return get_creating_sub_agent_info_impl(authorization)
+        return await get_creating_sub_agent_info_impl(authorization)
     except Exception as e:
         raise HTTPException(status_code=500, detail=f"Agent create error: {str(e)}")
 
@@ -68,7 +68,7 @@ async def update_agent_info_api(request: AgentInfoRequest, authorization: Option
     Update an existing agent
     """
     try:
-        update_agent_info_impl(request, authorization)
+        await update_agent_info_impl(request, authorization)
         return {}
     except Exception as e:
         raise HTTPException(status_code=500, detail=f"Agent update error: {str(e)}")
@@ -115,8 +115,8 @@ async def list_all_agent_info_api(authorization: Optional[str] = Header(None), r
     list all agent info
     """
     try:
-        user_id, tenant_id, _ = get_current_user_info(authorization, request)
-        return list_all_agent_info_impl(tenant_id=tenant_id, user_id=user_id)
+        _, tenant_id, _ = get_current_user_info(authorization, request)
+        return await list_all_agent_info_impl(tenant_id=tenant_id)
     except Exception as e:
         raise HTTPException(status_code=500, detail=f"Agent list error: {str(e)}")
 

backend/apps/northbound_app.py

@@ -0,0 +1,208 @@
+import logging
+from http import HTTPStatus
+from http.client import HTTPException
+from typing import Optional, Dict
+import uuid
+
+from fastapi import APIRouter, Body, Header, Request
+from fastapi.responses import JSONResponse
+
+from consts.exceptions import UnauthorizedError, LimitExceededError, SignatureValidationError
+from services.northbound_service import (
+    NorthboundContext,
+    get_conversation_history,
+    list_conversations,
+    start_streaming_chat,
+    stop_chat,
+    get_agent_info_list,
+    update_conversation_title
+)
+
+from utils.auth_utils import get_current_user_id, validate_aksk_authentication
+
+
+router = APIRouter(prefix="/nb/v1", tags=["northbound"])
+
+
+def _get_header(headers: Dict[str, str], name: str) -> Optional[str]:
+    for k, v in headers.items():
+        if k.lower() == name.lower():
+            return v
+    return None
+
+
+async def _parse_northbound_context(request: Request) -> NorthboundContext:
+    """
+    Build northbound context from headers.
+
+    - X-Access-Key: Access key for AK/SK authentication
+    - X-Timestamp: Timestamp for signature validation
+    - X-Signature: HMAC-SHA256 signature signed with secret key
+    - Authorization: Bearer <jwt>, jwt contains sub (user_id)
+    - X-Request-Id: optional, generated if not provided
+    """
+    # 1. Verify AK/SK signature
+    try:
+        # Get request body for signature verification
+        request_body = ""
+        if request.method in ["POST", "PUT", "PATCH"]:
+            try:
+                body_bytes = await request.body()
+                request_body = body_bytes.decode('utf-8') if body_bytes else ""
+            except Exception as e:
+                logging.warning(f"Cannot read request body for signature verification: {e}")
+                request_body = ""
+        
+        validate_aksk_authentication(request.headers, request_body)
+    except (UnauthorizedError, LimitExceededError, SignatureValidationError) as e:
+        raise e
+    except Exception:
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail="Internal Server Error: cannot parse northbound context")
+
+
+    # 2. Parse JWT token
+    auth_header = _get_header(request.headers, "Authorization")
+    if not auth_header:
+        raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: No authorization header found")
+
+    # Use auth_utils to parse JWT token
+    try:
+        user_id, tenant_id = get_current_user_id(auth_header)
+        
+        if not user_id:
+            raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: missing user_id in JWT token")
+        if not tenant_id:
+            raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: unregistered user_id in JWT token")
+            
+    except Exception:
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail="Internal Server Error: cannot parse JWT token")
+
+    request_id = _get_header(request.headers, "X-Request-Id") or str(uuid.uuid4())
+
+    return NorthboundContext(
+        request_id=request_id,
+        tenant_id=tenant_id,
+        user_id=str(user_id),
+        authorization=auth_header,
+    )
+
+
+@router.get("/health")
+async def health_check():
+    return {"status": "healthy", "service": "northbound-api"}
+
+
+@router.post("/chat/run")
+async def run_chat(
+    request: Request,
+    conversation_id: str = Body(..., embed=True),
+    agent_name: str = Body(..., embed=True),
+    query: str = Body(..., embed=True),
+    idempotency_key: Optional[str] = Header(None, alias="Idempotency-Key"),
+):
+    try:
+        ctx: NorthboundContext = await _parse_northbound_context(request)
+        return await start_streaming_chat(
+            ctx=ctx,
+            external_conversation_id=conversation_id,
+            agent_name=agent_name,
+            query=query,
+            idempotency_key=idempotency_key,
+        )
+    except UnauthorizedError:
+        raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: AK/SK authentication failed")
+    except LimitExceededError:
+        raise HTTPException(status_code=HTTPStatus.TOO_MANY_REQUESTS, detail="Too Many Requests: rate limit exceeded")
+    except SignatureValidationError:
+        raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: invalid signature")
+    except Exception:
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail="Internal Server Error")
+    
+
+
+@router.get("/chat/stop/{conversation_id}")
+async def stop_chat_stream(request: Request, conversation_id: str):
+    try:
+        ctx: NorthboundContext = await _parse_northbound_context(request)
+        return await stop_chat(ctx=ctx, external_conversation_id=conversation_id)
+    except UnauthorizedError:
+        raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: AK/SK authentication failed")
+    except LimitExceededError:
+        raise HTTPException(status_code=HTTPStatus.TOO_MANY_REQUESTS, detail="Too Many Requests: rate limit exceeded")
+    except SignatureValidationError:
+        raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: invalid signature")
+    except Exception:
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail="Internal Server Error")
+
+
+@router.get("/conversations/{conversation_id}")
+async def get_history(request: Request, conversation_id: str):
+    try:
+        ctx: NorthboundContext = await _parse_northbound_context(request)
+        return await get_conversation_history(ctx=ctx, external_conversation_id=conversation_id)
+    except UnauthorizedError:
+        raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: AK/SK authentication failed")
+    except LimitExceededError:
+        raise HTTPException(status_code=HTTPStatus.TOO_MANY_REQUESTS, detail="Too Many Requests: rate limit exceeded")
+    except SignatureValidationError:
+        raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: invalid signature")
+    except Exception:
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail="Internal Server Error")
+
+
+@router.get("/agents")
+async def list_agents(request: Request):
+    try:
+        ctx: NorthboundContext = await _parse_northbound_context(request)
+        return await get_agent_info_list(ctx=ctx)
+    except UnauthorizedError:
+        raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: AK/SK authentication failed")
+    except LimitExceededError:
+        raise HTTPException(status_code=HTTPStatus.TOO_MANY_REQUESTS, detail="Too Many Requests: rate limit exceeded")
+    except SignatureValidationError:
+        raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: invalid signature")
+    except Exception:
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail="Internal Server Error")
+
+
+@router.get("/conversations")
+async def list_convs(request: Request):
+    try:
+        ctx: NorthboundContext = await _parse_northbound_context(request)
+        return await list_conversations(ctx=ctx)
+    except UnauthorizedError:
+        raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: AK/SK authentication failed")
+    except LimitExceededError:
+        raise HTTPException(status_code=HTTPStatus.TOO_MANY_REQUESTS, detail="Too Many Requests: rate limit exceeded")
+    except SignatureValidationError:
+        raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: invalid signature")
+    except Exception:
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail="Internal Server Error")
+
+
+@router.put("/conversations/{conversation_id}/title")
+async def update_convs_title(
+    request: Request,
+    conversation_id: str,
+    title: str,
+    idempotency_key: Optional[str] = Header(None, alias="Idempotency-Key"),
+):
+    try:
+        ctx: NorthboundContext = await _parse_northbound_context(request)
+        result = await update_conversation_title(
+            ctx=ctx,
+            external_conversation_id=conversation_id,
+            title=title,
+            idempotency_key=idempotency_key,
+        )
+        headers_out = {"Idempotency-Key": result.get("idempotency_key", ""), "X-Request-Id": ctx.request_id}
+        return JSONResponse(content=result, headers=headers_out)
+
+    except UnauthorizedError:
+        raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: AK/SK authentication failed")
+    except LimitExceededError:
+        raise HTTPException(status_code=HTTPStatus.TOO_MANY_REQUESTS, detail="Too Many Requests: rate limit exceeded")
+    except SignatureValidationError:
+        raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized: invalid signature")
+    except Exception:
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail="Internal Server Error")

backend/apps/northbound_base_app.py

@@ -0,0 +1,46 @@
+from http import HTTPStatus
+import logging
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import JSONResponse
+from fastapi.middleware.cors import CORSMiddleware
+
+from .northbound_app import router as northbound_router
+from consts.exceptions import LimitExceededError, UnauthorizedError, SignatureValidationError
+
+logger = logging.getLogger("northbound_base_app")
+
+
+northbound_app = FastAPI(
+    title="Nexent Northbound API",
+    description="Northbound APIs for partners",
+    version="1.0.0",
+    root_path="/api"
+)
+
+northbound_app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["GET", "POST", "PUT", "DELETE"],
+    allow_headers=["*"],
+)
+
+
+northbound_app.include_router(northbound_router)
+
+
+@northbound_app.exception_handler(HTTPException)
+async def northbound_http_exception_handler(request, exc):
+    logger.error(f"Northbound HTTPException: {exc.detail}")
+    return JSONResponse(
+        status_code=exc.status_code,
+        content={"message": exc.detail},
+    )
+
+@northbound_app.exception_handler(Exception)
+async def northbound_generic_exception_handler(request, exc):
+    logger.error(f"Northbound Generic Exception: {exc}")
+    return JSONResponse(
+        status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+        content={"message": "Internal server error, please try again later."},
+    ) 

backend/apps/user_management_app.py

@@ -9,7 +9,7 @@
 from consts.const import SUPABASE_URL, SUPABASE_KEY
 from consts.model import STATUS_CODES, ServiceResponse, UserSignUpRequest, UserSignInRequest
 from database.model_management_db import create_model_record
-from utils.auth_utils import get_jwt_expiry_seconds, calculate_expires_at
+from utils.auth_utils import get_jwt_expiry_seconds, calculate_expires_at, get_current_user_id
 from database.user_tenant_db import insert_user_tenant
 from utils.config_utils import config_manager
 
@@ -550,8 +550,7 @@ async def get_user_id(request: Request):
 
     # If the token is invalid, try to parse the user ID from the token
     try:
-        from utils.auth_utils import get_current_user_id_from_token
-        user_id = get_current_user_id_from_token(authorization)
+        user_id, _ = get_current_user_id(authorization)
         if user_id:
             logging.info(f"Successfully parsed user ID from token: {user_id}")
             return ServiceResponse(

backend/consts/exceptions.py

@@ -0,0 +1,12 @@
+# Northbound customize error class
+class LimitExceededError(Exception):
+    """Raised when an outer platform calling too frequently"""
+    pass
+
+class UnauthorizedError(Exception):
+    """Raised when a user from outer platform is unauthorized."""
+    pass
+
+class SignatureValidationError(Exception):
+    """Raised when X-Signature header is missing or does not match the expected HMAC value."""
+    pass

backend/consts/model.py

@@ -378,4 +378,4 @@ class MemoryAgentShareMode(str, Enum):
 
     @classmethod
     def default(cls) -> "MemoryAgentShareMode":
-        return cls.NEVER
+        return cls.NEVER

backend/database/conversation_db.py

@@ -484,7 +484,7 @@ def get_conversation_history(conversation_id: int, user_id: Optional[str] = None
         ).where(
             ConversationMessageUnit.message_id == ConversationMessage.message_id,
             ConversationMessageUnit.delete_flag == 'N',
-            ConversationMessageUnit.unit_type != None
+            ConversationMessageUnit.unit_type is not None
         ).scalar_subquery()
 
         query = select(

backend/database/db_models.py

@@ -305,4 +305,24 @@ class AgentRelation(TableBase):
     update_time = Column(TIMESTAMP(timezone=False), server_default=func.now(), doc="Update time, audit field")
     created_by = Column(String(100), doc="Creator ID, audit field")
     updated_by = Column(String(100), doc="Last updater ID, audit field")
+    delete_flag = Column(String(1), default="N", doc="Delete flag, set to Y for soft delete, optional values Y/N")
+
+
+class PartnerMappingId(TableBase):
+    """
+    External-Internal ID mapping table for partners
+    """
+    __tablename__ = "partner_mapping_id_t"
+    __table_args__ = {"schema": SCHEMA}
+
+    mapping_id = Column(Integer, Sequence("partner_mapping_id_t_mapping_id_seq", schema=SCHEMA), primary_key=True, nullable=False, doc="ID")
+    external_id = Column(String(100), doc="The external id given by the outer partner")
+    internal_id = Column(Integer, doc="The internal id of the other database table")
+    mapping_type = Column(String(30), doc="Type of the external - internal mapping, value set: CONVERSATION")
+    tenant_id = Column(String(100), doc="Tenant ID")
+    user_id = Column(String(100), doc="User ID")
+    create_time = Column(TIMESTAMP(timezone=False), server_default=func.now(), doc="Creation time, audit field")
+    update_time = Column(TIMESTAMP(timezone=False), server_default=func.now(), doc="Update time, audit field")
+    created_by = Column(String(100), doc="Creator ID, audit field")
+    updated_by = Column(String(100), doc="Last updater ID, audit field")
     delete_flag = Column(String(1), default="N", doc="Delete flag, set to Y for soft delete, optional values Y/N")