Merge pull request #3687 from Agenta-AI/feat/refine-ai-feature

feat(api): AI services backend — refine prompt tool call

Author: junaway

AGENTS.md

@@ -204,6 +204,137 @@ async def create_workflow(
     ...
 ```
 
+### Domain-level exceptions
+
+1. **Define domain exceptions in the core layer** (`core/{domain}/types.py` or `core/{domain}/dtos.py`) — never raise `HTTPException` from services or DAOs.
+2. **Catch domain exceptions at the API boundary** — in the router or via a decorator — and convert them to HTTP responses.
+3. **Use a base exception per domain** so callers can catch broadly or narrowly.
+4. **Include structured context** (not just a message string) so the router can build rich HTTP error responses.
+
+**Example 1 — Folder exceptions (best example of the full pattern):**
+
+Definition in `api/oss/src/core/folders/types.py`:
+```python
+class FolderNameInvalid(Exception):
+    def __init__(self, message: str = "Folder name contains invalid characters."):
+        self.message = message
+        super().__init__(message)
+
+class FolderPathConflict(Exception):
+    def __init__(self, message: str = "A folder with this path already exists."):
+        self.message = message
+        super().__init__(message)
+```
+
+Raised in service `api/oss/src/core/folders/service.py`:
+```python
+def _validate_folder_name(name: Optional[str]) -> None:
+    if not name or not fullmatch(r"[\w -]+", name):
+        raise FolderNameInvalid()
+```
+
+Caught in router via decorator `api/oss/src/apis/fastapi/folders/router.py`:
+```python
+def handle_folder_exceptions():
+    def decorator(func):
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+            try:
+                return await func(*args, **kwargs)
+            except FolderNameInvalid as e:
+                raise FolderNameInvalidException(message=e.message) from e
+            except FolderPathConflict as e:
+                raise FolderPathConflictException(message=e.message) from e
+        return wrapper
+    return decorator
+```
+
+**Example 2 — Filtering exception (inline catch in router):**
+
+Definition in `api/oss/src/core/tracing/dtos.py`:
+```python
+class FilteringException(Exception):
+    pass
+```
+
+Caught in `api/oss/src/apis/fastapi/tracing/router.py`:
+```python
+except FilteringException as e:
+    raise HTTPException(status_code=400, detail=str(e)) from e
+```
+
+**Example 3 — EE organization exceptions (base class hierarchy):**
+
+Definition in `api/ee/src/core/organizations/exceptions.py`:
+```python
+class OrganizationError(Exception):
+    """Base exception for organization-related errors."""
+    pass
+
+class OrganizationSlugConflictError(OrganizationError):
+    def __init__(self, slug: str, message: str = None):
+        self.slug = slug
+        self.message = message or f"Organization slug '{slug}' is already in use."
+        super().__init__(self.message)
+
+class OrganizationNotFoundError(OrganizationError):
+    def __init__(self, organization_id: str, message: str = None):
+        self.organization_id = organization_id
+        self.message = message or f"Organization '{organization_id}' not found."
+        super().__init__(self.message)
+```
+
+Anti-patterns:
+- Do NOT return error dicts like `{"_error": True, "status_code": 502, "detail": "..."}` from services or clients.
+- Do NOT raise `HTTPException` from core services — that couples the domain to HTTP.
+- Do NOT use bare `Exception` or `ValueError` for domain errors when a typed exception would be clearer.
+
+### Typed DTO returns from services and clients
+
+1. **Service methods must return typed DTOs** (Pydantic `BaseModel` subclasses), not raw dicts, tuples, or `Any`.
+2. **HTTP clients should return a response DTO**, not `Tuple[Optional[Any], Optional[str]]`.
+3. **Define DTOs in `core/{domain}/dtos.py`** alongside the domain's other data contracts.
+4. **Use `Optional[DTO]` for missing entities**, `List[DTO]` for collections.
+
+**Example — Workflow service returns typed DTOs:**
+
+```python
+# core/workflows/dtos.py
+class Workflow(Artifact):
+    pass
+
+class WorkflowVariant(Variant):
+    pass
+
+# core/workflows/service.py
+async def create_workflow(self, *, project_id, user_id, workflow_create) -> Optional[Workflow]:
+    artifact = await self.workflows_dao.create_artifact(...)
+    if not artifact:
+        return None
+    return Workflow(**artifact.model_dump(mode="json"))
+```
+
+**Example — HTTP client returns a DTO instead of a raw tuple:**
+
+```python
+# Instead of:
+async def invoke(...) -> Tuple[Optional[Any], Optional[str]]:
+    return data, trace_id  # untyped
+
+# Do:
+class InvokeResponse(BaseModel):
+    data: Any
+    trace_id: Optional[str] = None
+
+async def invoke(...) -> InvokeResponse:
+    return InvokeResponse(data=data, trace_id=trace_id)
+```
+
+Anti-patterns:
+- Do NOT return raw dicts from service methods.
+- Do NOT return tuples like `(data, trace_id)` — use a named DTO.
+- Do NOT use `Dict[str, Any]` as a return type when you can define a proper model.
+
 ### Migration seams (do not copy for net-new code)
 
 These exist during transition but should not be copied into new implementations:

api/ee/src/core/entitlements/types.py

@@ -75,6 +75,7 @@ class Category(str, Enum):
     TRACING_SLOW = "tracing_slow"
     SERVICES_FAST = "services_fast"
     SERVICES_SLOW = "services_slow"
+    AI_SERVICES = "ai_services"
 
 
 class Method(str, Enum):
@@ -122,6 +123,9 @@ class Throttle(BaseModel):
     Category.SERVICES_SLOW: [
         # None defined yet
     ],
+    Category.AI_SERVICES: [
+        (Method.POST, "/ai/services/tools/call"),
+    ],
     Category.STANDARD: [
         # None defined yet
         # CATCH ALL
@@ -364,6 +368,16 @@ class Throttle(BaseModel):
                     rate=1,
                 ),
             ),
+            Throttle(
+                categories=[
+                    Category.AI_SERVICES,
+                ],
+                mode=Mode.INCLUDE,
+                bucket=Bucket(
+                    capacity=10,
+                    rate=30,
+                ),
+            ),
         ],
     },
     Plan.CLOUD_V0_PRO: {
@@ -436,6 +450,16 @@ class Throttle(BaseModel):
                     rate=1,
                 ),
             ),
+            Throttle(
+                categories=[
+                    Category.AI_SERVICES,
+                ],
+                mode=Mode.INCLUDE,
+                bucket=Bucket(
+                    capacity=30,
+                    rate=90,
+                ),
+            ),
         ],
     },
     Plan.CLOUD_V0_BUSINESS: {
@@ -506,6 +530,16 @@ class Throttle(BaseModel):
                     rate=1,
                 ),
             ),
+            Throttle(
+                categories=[
+                    Category.AI_SERVICES,
+                ],
+                mode=Mode.INCLUDE,
+                bucket=Bucket(
+                    capacity=300,
+                    rate=900,
+                ),
+            ),
         ],
     },
     Plan.CLOUD_V0_HUMANITY_LABS: {

api/entrypoints/routers.py

@@ -103,6 +103,9 @@
 from oss.src.apis.fastapi.evaluations.router import EvaluationsRouter
 from oss.src.apis.fastapi.evaluations.router import SimpleEvaluationsRouter
 
+from oss.src.core.ai_services.service import AIServicesService
+from oss.src.apis.fastapi.ai_services.router import AIServicesRouter
+
 
 from oss.src.routers import (
     admin_router,
@@ -431,6 +434,13 @@ async def lifespan(*args, **kwargs):
     annotations_service=annotations_service,
 )
 
+# AI SERVICES ------------------------------------------------------------------
+
+ai_services_service = AIServicesService.from_env()
+ai_services = AIServicesRouter(
+    ai_services_service=ai_services_service,
+)
+
 # MOUNTING ROUTERS TO APP ROUTES -----------------------------------------------
 
 app.include_router(
@@ -532,6 +542,12 @@ async def lifespan(*args, **kwargs):
     tags=["Workflows"],
 )
 
+app.include_router(
+    router=ai_services.router,
+    prefix="/ai/services",
+    tags=["AI Services"],
+)
+
 app.include_router(
     router=evaluators.router,
     prefix="/preview/evaluators",

api/oss/src/apis/fastapi/ai_services/__init__.py

@@ -0,0 +1 @@
+

api/oss/src/apis/fastapi/ai_services/models.py

@@ -0,0 +1,17 @@
+from oss.src.core.ai_services.dtos import (
+    AIServicesStatus,
+    ToolCallRequest,
+    ToolCallResponse,
+)
+
+
+class AIServicesStatusResponse(AIServicesStatus):
+    pass
+
+
+class ToolCallRequestModel(ToolCallRequest):
+    pass
+
+
+class ToolCallResponseModel(ToolCallResponse):
+    pass

api/oss/src/apis/fastapi/ai_services/router.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, HTTPException, status
+from pydantic import ValidationError
+
+from oss.src.utils.exceptions import intercept_exceptions
+
+from oss.src.core.ai_services.dtos import AIServicesUnknownToolError
+from oss.src.core.ai_services.service import AIServicesService
+from oss.src.apis.fastapi.ai_services.models import (
+    AIServicesStatusResponse,
+    ToolCallRequestModel,
+    ToolCallResponseModel,
+)
+
+
+class AIServicesRouter:
+    def __init__(
+        self,
+        *,
+        ai_services_service: AIServicesService,
+    ):
+        self.service = ai_services_service
+        self.router = APIRouter()
+
+        self.router.add_api_route(
+            "/status",
+            self.get_status,
+            methods=["GET"],
+            operation_id="ai_services_status",
+            status_code=status.HTTP_200_OK,
+            response_model=AIServicesStatusResponse,
+            response_model_exclude_none=True,
+        )
+
+        self.router.add_api_route(
+            "/tools/call",
+            self.call_tool,
+            methods=["POST"],
+            operation_id="ai_services_tools_call",
+            status_code=status.HTTP_200_OK,
+            response_model=ToolCallResponseModel,
+            response_model_exclude_none=True,
+        )
+
+    @intercept_exceptions()
+    async def get_status(self) -> AIServicesStatusResponse:
+        # TODO: Access control should be org-level feature flag (org owner
+        # enables/disables AI services for the whole org) rather than
+        # per-user permissions.  For now, env-var gating is sufficient.
+        return self.service.status()
+
+    @intercept_exceptions()
+    async def call_tool(
+        self,
+        *,
+        tool_call: ToolCallRequestModel,
+    ) -> ToolCallResponseModel:
+        if not self.service.enabled:
+            raise HTTPException(status_code=503, detail="AI services are disabled")
+
+        # TODO: Access control should be org-level feature flag (org owner
+        # enables/disables AI services for the whole org) rather than
+        # per-user permissions.  For now, env-var gating is sufficient.
+
+        try:
+            return await self.service.call_tool(
+                name=tool_call.name,
+                arguments=tool_call.arguments,
+            )
+        except AIServicesUnknownToolError as e:
+            raise HTTPException(status_code=400, detail=e.message) from e
+        except ValidationError as e:
+            raise HTTPException(status_code=400, detail=e.errors()) from e

api/oss/src/core/ai_services/__init__.py

@@ -0,0 +1 @@
+