ai

ai

AICRUDError

AICRUDError(
    operation: str,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when AI service create, update, or delete operations fail.

Source code in src/crew_dcs/routes/ai.py

def __init__(
    self,
    operation: str,
    message: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message or f"AI service {operation} failed",
        res=res,
        **kwargs,
    )

AIGETError

AIGETError(message: str | None = None, res=None, **kwargs)

Bases: RouteError

Raised when AI service retrieval operations fail.

Source code in src/crew_dcs/routes/ai.py

def __init__(self, message: str | None = None, res=None, **kwargs):
    super().__init__(
        message=message or "AI service retrieval failed",
        res=res,
        **kwargs,
    )

generate_chat_body

generate_chat_body(
    text_input: str,
    model: str = "domo.domo_ai.domogpt-chat-medium-v1.1:anthropic",
) -> dict

Build a /text/generation body.

WARNING: model availability is PER-INSTANCE. This default 404s on datacrew-space (DS-0043 "ML Model does not exist or access denied"); domo.domo_ai.domogpt-medium-v2.2:anthropic works there. Resolve the model at runtime — never ship a hardcoded default to another tenant.

Structured output: add responseFormat (spec: JsonResponseFormat in /openapi/product/AI-Services.yaml)::

body["responseFormat"] = {"type": "JSON", "schema": {...json schema...}}

type is UPPERCASE "JSON". Confirmed by elimination — json, json_schema, JSON_OBJECT, object all 400. Returns 200 with choices[0].output as a schema-conforming JSON string.

Source code in src/crew_dcs/routes/ai.py

def generate_chat_body(
    text_input: str, model: str = "domo.domo_ai.domogpt-chat-medium-v1.1:anthropic"
) -> dict:
    """Build a `/text/generation` body.

    WARNING: model availability is PER-INSTANCE. This default 404s on
    `datacrew-space` (`DS-0043 "ML Model does not exist or access denied"`);
    `domo.domo_ai.domogpt-medium-v2.2:anthropic` works there. Resolve the model
    at runtime — never ship a hardcoded default to another tenant.

    Structured output: add `responseFormat` (spec: `JsonResponseFormat` in
    `/openapi/product/AI-Services.yaml`)::

        body["responseFormat"] = {"type": "JSON", "schema": {...json schema...}}

    `type` is UPPERCASE `"JSON"`. Confirmed by elimination — `json`,
    `json_schema`, `JSON_OBJECT`, `object` all 400. Returns 200 with
    `choices[0].output` as a schema-conforming JSON string.
    """
    return {
        "input": text_input,
        "promptTemplate": {"template": "${input}"},
        "model": model,
    }

generate_image_to_text_body

generate_image_to_text_body(
    image_data: str,
    media_type: str = "image/png",
    text_input: str | None = None,
    system_prompt: str | None = None,
    model: str = "domo.domo_ai.domogpt-chat-medium-v1.1:anthropic",
) -> dict

Build the request body for the Image to Text AI service.

Source code in src/crew_dcs/routes/ai.py

def generate_image_to_text_body(
    image_data: str,
    media_type: str = "image/png",
    text_input: str | None = None,
    system_prompt: str | None = None,
    model: str = "domo.domo_ai.domogpt-chat-medium-v1.1:anthropic",
) -> dict:
    """Build the request body for the Image to Text AI service."""
    default_system = (
        "You are an AI assistant tasked with performing Optical Character Recognition "
        "(OCR) on an image. Your goal is to accurately identify and transcribe any text "
        "present in the image."
    )
    return {
        "input": text_input or "",
        "system": system_prompt or default_system,
        "promptTemplate": {"template": "${input}"},
        "model": model,
        "image": {
            "mediaType": media_type,
            "type": "base64",
            "data": image_data,
        },
    }

llm_generate_text async

llm_generate_text(
    text_input: str,
    auth: DomoAuth,
    chat_body: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

POST /api/ai/v1/text/generation.

Pass chat_body to override the default (see generate_chat_body for the per-instance model trap and the responseFormat structured-output shape).

Debug tip — the status code tells you which half is wrong: 400 = field name recognised, value rejected 404 = field name unrecognised, fell through to model lookup (DS-0043)

Source code in src/crew_dcs/routes/ai.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def llm_generate_text(
    text_input: str,
    auth: DomoAuth,
    chat_body: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """POST /api/ai/v1/text/generation.

    Pass `chat_body` to override the default (see `generate_chat_body` for the
    per-instance model trap and the `responseFormat` structured-output shape).

    Debug tip — the status code tells you which half is wrong:
      400 = field name recognised, value rejected
      404 = field name unrecognised, fell through to model lookup (`DS-0043`)
    """
    url = f"https://{auth.domo_instance}.domo.com/api/ai/v1/text/generation"

    body = chat_body or generate_chat_body(text_input=text_input)

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=body,
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise AICRUDError(operation="generate text", res=res)

    res.response["output"] = res.response["choices"][0]["output"]

    return res

run_image_to_text async

run_image_to_text(
    auth: DomoAuth,
    image_data: str,
    media_type: str = "image/png",
    text_input: str | None = None,
    system_prompt: str | None = None,
    model: str = "domo.domo_ai.domogpt-chat-medium-v1.1:anthropic",
    body: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Run the Domo AI Image to Text (OCR) model.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Domo authentication credentials.	required
image_data	str	Base64-encoded image content.	required
media_type	str	MIME type of the image (e.g. "image/png").	'image/png'
text_input	str | None	Optional additional instructions to pass as input.	None
system_prompt	str | None	Override the default OCR system prompt.	None
model	str	Domo AI model identifier.	'domo.domo_ai.domogpt-chat-medium-v1.1:anthropic'
body	dict | None	Override the entire request body (all other args ignored).	None
return_raw	bool	Return raw ResponseGetData without extracting text.	False

Source code in src/crew_dcs/routes/ai.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def run_image_to_text(
    auth: DomoAuth,
    image_data: str,
    media_type: str = "image/png",
    text_input: str | None = None,
    system_prompt: str | None = None,
    model: str = "domo.domo_ai.domogpt-chat-medium-v1.1:anthropic",
    body: dict | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Run the Domo AI Image to Text (OCR) model.

    Args:
        auth: Domo authentication credentials.
        image_data: Base64-encoded image content.
        media_type: MIME type of the image (e.g. ``"image/png"``).
        text_input: Optional additional instructions to pass as input.
        system_prompt: Override the default OCR system prompt.
        model: Domo AI model identifier.
        body: Override the entire request body (all other args ignored).
        return_raw: Return raw ``ResponseGetData`` without extracting text.
    """
    url = f"https://{auth.domo_instance}.domo.com/api/ai/v1/image/text"

    request_body = body or generate_image_to_text_body(
        image_data=image_data,
        media_type=media_type,
        text_input=text_input,
        system_prompt=system_prompt,
        model=model,
    )

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="POST",
        body=request_body,
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise AICRUDError(operation="image to text", res=res)

    res.response["output"] = res.response["choices"][0]["output"]

    return res