Api reference

components.ai_tooling.public.actions ¶

review_response ¶

Actions for submitting AI review responses.

submit_review_response ¶

submit_review_response(
    entity_id, reviewer_id, answers, agent_type, review_type
)

Submit a review response for an AI conversation.

Finds the latest form version for the agent_type/review_type combination and stores the reviewer's answers.

Parameters:

Name	Type	Description	Default
`entity_id`	`UUID`	The conversation or response ID being reviewed.	required
`reviewer_id`	`str`	"Local" user id of the Alaner submitting the review (we then get the global id).	required
`answers`	`dict[str, Any]`	Mapping of question_key to answer_value.	required
`agent_type`	`str`	The type of AI agent being reviewed.	required
`review_type`	`ReviewType`	The type of review workflow.	required

Returns:

Type	Description
`ReviewResponseResult`	A ReviewResponseResult with the created response data.

Raises:

Type	Description
`missing_resource`	If no form exists for the agent_type/review_type.

Source code in components/ai_tooling/public/actions/review_response.py

def submit_review_response(
    entity_id: UUID,
    reviewer_id: str,
    answers: dict[str, Any],
    agent_type: str,
    review_type: ReviewType,
) -> ReviewResponseResult:
    """Submit a review response for an AI conversation.

    Finds the latest form version for the agent_type/review_type combination
    and stores the reviewer's answers.

    Args:
        entity_id: The conversation or response ID being reviewed.
        reviewer_id: "Local" user id of the Alaner submitting the review (we then get the global id).
        answers: Mapping of question_key to answer_value.
        agent_type: The type of AI agent being reviewed.
        review_type: The type of review workflow.

    Returns:
        A ReviewResponseResult with the created response data.

    Raises:
        BaseErrorCode.missing_resource: If no form exists for the agent_type/review_type.
    """
    user_profile = ProfileService.create().get_user_profile(user_id=reviewer_id)
    if user_profile:
        global_reviewer_id = user_profile.global_user_id
    else:
        current_logger.warning(
            f"Profile could not be found for user with id {reviewer_id}. Submitting AI review form with user id instead."
        )
        global_reviewer_id = reviewer_id
    stmt = (
        select(AIReviewForm)
        .where(
            AIReviewForm.agent_type == agent_type,
            AIReviewForm.review_type == review_type.value,
        )
        .order_by(AIReviewForm.version.desc())
        .limit(1)
    )
    form = current_session.execute(stmt).scalar_one_or_none()

    if form is None:
        raise BaseErrorCode.missing_resource(
            f"No review form found for agent_type={agent_type}, review_type={review_type.value}"
        )

    response = AIReviewResponse(
        form_id=form.id,
        entity_id=entity_id,
        agent_type=agent_type,
        review_type=review_type,
        reviewer_id=global_reviewer_id,
        answers=answers,
    )

    current_session.add(response)
    current_session.commit()

    return ReviewResponseResult(
        id=response.id,
        entity_id=response.entity_id,
        reviewer_id=response.reviewer_id,
        agent_type=response.agent_type,
        answers=response.answers,
    )

components.ai_tooling.public.agent_config_resolver ¶

Resolve the (env, branch, version) triple for Marmot agent prompt configs.

Originally specific to the AR store; now accepts any S3ConfigStore via the optional store= parameter, so other agents can resolve their own branch/version against dedicated stores (see ai_tooling_service.AiToolingService).

get_current_user_branch ¶

get_current_user_branch()

Return the developer's IAM user on dev, else main.

Source code in components/ai_tooling/public/agent_config_resolver.py

def get_current_user_branch() -> str:
    """Return the developer's IAM user on dev, else `main`."""
    if is_development_mode():
        current_iam_user = get_current_user()
        if current_iam_user:
            return current_iam_user

    return "main"

get_latest_version_from_s3 ¶

get_latest_version_from_s3(branch, env=None, store=None)

Return the latest version folder for (env, branch) from S3 commit history.

Source code in components/ai_tooling/public/agent_config_resolver.py

def get_latest_version_from_s3(
    branch: str, env: str | None = None, store: S3ConfigStore | None = None
) -> str:
    """Return the latest version folder for `(env, branch)` from S3 commit history."""
    if not env:
        env = get_env_name()
    if store is None:
        store = _fr_ar_config_store

    history = store.fetch_commit_history(env, branch)
    if not history:
        raise Exception(f"No commit history found for env={env} branch={branch}")
    return history[-1]

get_nonnull_env_branch_version ¶

get_nonnull_env_branch_version(
    env=None, branch=None, version=None, store=None
)

Resolve (env, branch, version) from potentially-null inputs.

Defaults: env from runtime, branch from current IAM user (dev) / main, version from latest S3 commit. On dev, falls back to main if the user's branch has no commit history yet.

If store is None, defaults to the AR FR-scoped store (legacy behavior).

Source code in components/ai_tooling/public/agent_config_resolver.py

def get_nonnull_env_branch_version(
    env: str | None = None,
    branch: str | None = None,
    version: str | None = None,
    store: S3ConfigStore | None = None,
) -> tuple[str, str, str]:
    """Resolve `(env, branch, version)` from potentially-null inputs.

    Defaults: `env` from runtime, `branch` from current IAM user (dev) / `main`,
    `version` from latest S3 commit. On dev, falls back to `main` if the user's
    branch has no commit history yet.

    If `store` is None, defaults to the AR FR-scoped store (legacy behavior).
    """
    if not env:
        env = get_env_name()
    if not branch:
        branch = get_current_user_branch()
    if store is None:
        store = _fr_ar_config_store

    try:
        if not version:
            version = get_latest_version_from_s3(branch=branch, env=env, store=store)
        return env, branch, version
    except Exception:
        if env == "dev" and branch != "main":
            current_logger.info(
                f"User branch '{branch}' not found in S3, falling back to main branch"
            )
            return get_nonnull_env_branch_version(
                env=env, branch="main", version=version, store=store
            )
        raise

components.ai_tooling.public.agent_config_stores ¶

ar_config_store `module-attribute` ¶

ar_config_store = S3ConfigStore(
    prefix="automated-resolution",
    sections=[agents, llm_prompt_templates, redirections],
)

harry_config_store `module-attribute` ¶

harry_config_store = S3ConfigStore(
    prefix="harry", sections=[agents, llm_prompt_templates]
)

legal_complaint_config_store `module-attribute` ¶

legal_complaint_config_store = S3ConfigStore(
    prefix="legal-complaint",
    sections=[agents, llm_prompt_templates],
)

components.ai_tooling.public.ai_agent_config_tool ¶

AIAgentConfigTool ¶

AIAgentConfigTool(
    config_store,
    view_permissions,
    edit_permissions,
    enrichers=(),
    review_notifications_slack_channel=None,
    read_only=False,
)

Configurable agent configuration tool.

Parameters:

Name	Type	Description	Default
`config_store`	`ConfigStore`	Config store (pure data access).	required
`enrichers`	`Sequence[ConfigEnricher]`	Transforms applied in order to fetched configs before they're returned. The store fetches raw data; each consumer declares its own enrichment at the composition root (e.g. `standard_sections_enricher`).	`()`
`view_permissions`	`set[EmployeePermission]`	Employee permissions required for read endpoints.	required
`edit_permissions`	`set[EmployeePermission]`	Employee permissions required for write endpoints.	required
`review_notifications_slack_channel`	`str \| None`	Slack channel for review notifications.	`None`
`read_only`	`bool`	When True, only register read endpoints (the write endpoints are skipped).	`False`

Source code in components/ai_tooling/public/ai_agent_config_tool.py

def __init__(
    self,
    config_store: ConfigStore,
    view_permissions: set[EmployeePermission],
    edit_permissions: set[EmployeePermission],
    enrichers: Sequence[ConfigEnricher] = (),
    review_notifications_slack_channel: str | None = None,
    read_only: bool = False,
) -> None:
    self._config_store = config_store
    self._enrichers = enrichers
    self._view_permissions = view_permissions
    self._edit_permissions = edit_permissions
    self._review_notifications_slack_channel = review_notifications_slack_channel
    self._read_only = read_only

get_llm_prompt_template ¶

get_llm_prompt_template(
    prompt_name,
    config_branch=None,
    config_version=None,
    config_env=None,
    app_name=None,
)

Return the raw prompt template dict.

Source code in components/ai_tooling/public/ai_agent_config_tool.py

def get_llm_prompt_template(
    self,
    prompt_name: str,
    config_branch: str | None = None,
    config_version: str | None = None,
    config_env: str | None = None,
    app_name: AppName | None = None,
) -> dict[str, Any]:
    """Return the raw prompt template dict."""
    # Prompt-template lookup relies on S3 commit history; it's only used with
    # S3-backed stores (AR/Harry/legal/FR), never the git-backed ops store.
    store = cast("S3ConfigStore", self._config_store.for_country(app_name))

    env, branch, version = get_nonnull_env_branch_version(
        env=config_env, branch=config_branch, version=config_version, store=store
    )
    templates_data = store.fetch_section(
        ConfigSection.llm_prompt_templates.value, env, branch, version
    )
    template_data: dict[str, Any] | None = next(
        (t for t in templates_data if t.get("name") == prompt_name), None
    )
    if template_data is None:
        raise BaseErrorCode.missing_resource(
            f"LLM prompt template '{prompt_name}' not found "
            f"(prefix='{store.prefix}' branch={branch} version={version})"
        )
    return template_data

get_member_attribute_names ¶

get_member_attribute_names(
    agent_name,
    config_branch=None,
    config_version=None,
    config_env=None,
    app_name=None,
)

Return the member-attribute names configured for an agent.

Source code in components/ai_tooling/public/ai_agent_config_tool.py

def get_member_attribute_names(
    self,
    agent_name: str,
    config_branch: str | None = None,
    config_version: str | None = None,
    config_env: str | None = None,
    app_name: AppName | None = None,
) -> list[str]:
    """Return the member-attribute names configured for an agent."""
    # Member-attribute config is only stored in S3-backed stores (AR/Harry/FR).
    store = cast("S3ConfigStore", self._config_store.for_country(app_name))

    env, branch, version = get_nonnull_env_branch_version(
        env=config_env, branch=config_branch, version=config_version, store=store
    )
    agents = store.fetch_section(ConfigSection.agents.value, env, branch, version)
    for agent in agents or []:
        if agent.get("name") == agent_name:
            return list(agent.get("member_attribute_names") or ())
    return []

register_smorest_routes ¶

register_smorest_routes(blueprint, prefix='')

Register agent config tool routes on a flask-smorest blueprint.

Endpoints (write endpoints are skipped when read_only): GET {prefix}/agent_config/all GET {prefix}/agent_config/branches GET {prefix}/agent_config/commit_history POST {prefix}/agent_config/save (write) POST {prefix}/agent_config/copy (write) POST {prefix}/agent_config/rollback (write) POST {prefix}/agent_config/delete_branch (write) POST {prefix}/agent_config/request_review (write) GET {prefix}/agent_config/review_state POST {prefix}/agent_config/approve_review (write) GET {prefix}/agent_config/agents GET {prefix}/agent_config/agent GET {prefix}/agent_config/tools

Source code in components/ai_tooling/public/ai_agent_config_tool.py

def register_smorest_routes(
    self, blueprint: CustomBlueprint, prefix: str = ""
) -> None:
    """Register agent config tool routes on a flask-smorest blueprint.

    Endpoints (write endpoints are skipped when `read_only`):
        GET  {prefix}/agent_config/all
        GET  {prefix}/agent_config/branches
        GET  {prefix}/agent_config/commit_history
        POST {prefix}/agent_config/save            (write)
        POST {prefix}/agent_config/copy            (write)
        POST {prefix}/agent_config/rollback        (write)
        POST {prefix}/agent_config/delete_branch   (write)
        POST {prefix}/agent_config/request_review  (write)
        GET  {prefix}/agent_config/review_state
        POST {prefix}/agent_config/approve_review  (write)
        GET  {prefix}/agent_config/agents
        GET  {prefix}/agent_config/agent
        GET  {prefix}/agent_config/tools
    """
    prefix = prefix.rstrip("/")
    view_policy = BackofficePermissionAccessPolicy.permitted_for(
        self._view_permissions
    )
    edit_policy = BackofficePermissionAccessPolicy.permitted_for(
        self._edit_permissions
    )

    def write_route(
        rule: str, **kwargs: Any
    ) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
        """`blueprint.route` for write endpoints; a no-op decorator
        (route left unregistered) when the tool is read-only.
        """
        if self._read_only:
            return lambda fn: fn
        return blueprint.route(rule, **kwargs)

    @blueprint.route(f"{prefix}/agent_config/all", methods=["GET"])
    @enforce_policy(view_policy)
    @blueprint.arguments(_GetAllGetQuerySchema, location="query")
    @blueprint.response(200)
    @obs.api_call()
    def _get_all(query_args: dict[str, str]) -> flask.Response:
        from components.ai_tooling.internal.agent_config_tool.business_logic.version_management import (
            resolve_env_branch_version,
        )

        try:
            store = self._country_store()
            env, branch, version = resolve_env_branch_version(
                store=store,
                env=query_args.get("env"),
                branch=query_args.get("branch"),
                version=query_args.get("version"),
            )

            configs = store.fetch_all(env, branch, version)
            for enrich in self._enrichers:
                configs = enrich(configs)
        except Exception as e:
            abort(400, message=str(e))
        return make_json_response(configs)

    @blueprint.route(f"{prefix}/agent_config/branches", methods=["GET"])
    @enforce_policy(view_policy)
    @blueprint.arguments(_GetBranchesGetQuerySchema, location="query")
    @blueprint.response(200)
    @obs.api_call()
    def _get_branches(query_args: dict[str, str]) -> flask.Response:
        env = query_args.get("env") or get_env_name()
        branches = self._country_store().list_branches(env)
        return make_json_response(branches)

    @blueprint.route(f"{prefix}/agent_config/commit_history", methods=["GET"])
    @enforce_policy(view_policy)
    @blueprint.arguments(_GetCommitHistoryGetQuerySchema, location="query")
    @blueprint.response(200)
    @obs.api_call()
    def _get_commit_history(query_args: dict[str, Any]) -> flask.Response:
        from components.ai_tooling.internal.agent_config_tool.business_logic.version_management import (
            resolve_env_branch_version,
        )

        try:
            store = self._country_store()
            env, branch, version = resolve_env_branch_version(
                store=store,
                branch=query_args.get("branch"),
                version=query_args.get("version"),
            )
            commits = store.fetch_recent_commits(
                env, branch, version, limit=query_args["limit"]
            )
        except Exception as e:
            abort(400, message=str(e))
        return make_json_response({"commits": [c.to_dict() for c in commits]})

    @write_route(f"{prefix}/agent_config/save", methods=["POST"])
    @enforce_policy(edit_policy)
    @blueprint.arguments(_SavePostJsonSchema, location="json")
    @blueprint.response(200)
    @obs.api_call()
    def _save(json_args: dict[str, Any]) -> flask.Response:
        from components.ai_tooling.internal.agent_config_tool.business_logic.version_management import (
            save_config,
        )

        env = json_args["env"]
        backend_env = get_env_name()
        if env != backend_env:
            abort(
                400,
                message=f"Environment ({env}) does not match server environment ({backend_env})",
            )

        store = self._country_store()
        save_config(store, env, json_args["branch"], json_args["changes"])
        return make_success_json_response()

    @write_route(f"{prefix}/agent_config/copy", methods=["POST"])
    @enforce_policy(edit_policy)
    @blueprint.arguments(_CopyPostJsonSchema, location="json")
    @blueprint.response(200)
    @obs.api_call()
    def _copy(json_args: dict[str, Any]) -> flask.Response:
        from components.ai_tooling.internal.agent_config_tool.business_logic.review_management import (
            notify_force_merge,
            verify_review_approved,
        )
        from components.ai_tooling.internal.agent_config_tool.business_logic.version_management import (
            get_default_branch,
            resolve_env_branch_version,
        )
        from components.ai_tooling.public.entities.agent_config import (
            AgentConfigCommitMetadata,
        )

        store = self._country_store()
        source_branch = json_args["source_branch"]
        source_env = json_args.get("source_env")
        target_env = json_args.get("target_env")
        is_force = bool(json_args.get("force"))
        target_branch = json_args.get("target_branch") or get_default_branch()
        commit_message = json_args.get("commit_message") or ""
        # Peer review only gates merges of a feature branch into main; other
        # copy flows (prod→env sync, main→feature fork, reset-from-main) are
        # not ships and don't require approval.
        is_ship_to_main = target_branch == "main" and source_branch != "main"
        peer_review_required = is_ship_to_main and self._is_peer_review_enabled()
        review_skipped = is_force and peer_review_required
        metadata = AgentConfigCommitMetadata.create(
            commit_message, review_skipped=review_skipped
        )

        try:
            if is_force or not peer_review_required:
                _, _, source_version = resolve_env_branch_version(
                    store, env=source_env, branch=source_branch
                )
            else:
                source_version = verify_review_approved(
                    store, env=source_env, branch=source_branch
                )
            store.copy(
                source_env=source_env,
                source_branch=source_branch,
                target_env=target_env,
                target_branch=target_branch,
                metadata=metadata,
                source_version=source_version,
            )
        except Exception as e:
            abort(400, message=str(e))

        # Stage's main is the source of truth for prod's main. When shipping
        # a feature branch to stage main also propagate to prod main.
        if is_ship_to_main and source_env == "stage" and target_env == "stage":
            try:
                store.copy(
                    source_env="stage",
                    source_branch="main",
                    target_env="prod",
                    target_branch=target_branch,
                    metadata=metadata,
                    # Use the version just written to stage/main
                    source_version=metadata.created_at,
                )
            except Exception as e:
                abort(
                    500,
                    message=f"Shipped to stage main but prod propagation failed: {e}. "
                    "Re-run ship to retry.",
                )

        if review_skipped and self._review_notifications_slack_channel:
            notify_force_merge(
                branch=source_branch,
                commit_message=commit_message,
                review_notifications_slack_channel=self._review_notifications_slack_channel,
            )

        return make_success_json_response()

    @write_route(f"{prefix}/agent_config/rollback", methods=["POST"])
    @enforce_policy(edit_policy)
    @blueprint.arguments(_RollbackPostJsonSchema, location="json")
    @blueprint.response(200)
    @obs.api_call()
    def _rollback(json_args: dict[str, str]) -> flask.Response:
        from components.ai_tooling.internal.agent_config_tool.business_logic.version_management import (
            get_default_branch,
        )
        from components.ai_tooling.public.entities.agent_config import (
            AgentConfigCommitMetadata,
        )

        env = get_env_name()
        branch = json_args.get("branch") or get_default_branch()
        target_version = json_args["target_version"]
        commit_message = f"Rollback to {target_version}"
        store = self._country_store()
        try:
            store.rollback(
                target_version=target_version,
                env=env,
                branch=branch,
                metadata=AgentConfigCommitMetadata.create(commit_message),
            )
        except Exception as e:
            abort(400, message=str(e))

        # Stage's main is the source of truth for prod's main. Propagate
        # the rollback so the two envs stay aligned.
        if env == "stage" and branch == "main":
            try:
                store.copy(
                    source_env="stage",
                    source_branch="main",
                    target_env="prod",
                    target_branch="main",
                    metadata=AgentConfigCommitMetadata.create(commit_message),
                )
            except Exception as e:
                abort(
                    500,
                    message=f"Rolled back stage to {target_version} but prod propagation failed: {e}. "
                    "Re-run rollback to retry.",
                )
        return make_success_json_response()

    @write_route(f"{prefix}/agent_config/delete_branch", methods=["POST"])
    @enforce_policy(edit_policy)
    @blueprint.arguments(_DeleteBranchPostJsonSchema, location="json")
    @blueprint.response(200)
    @obs.api_call()
    def _delete_branch(json_args: dict[str, str]) -> flask.Response:
        try:
            self._country_store().delete_branch(
                env=get_env_name(), branch=json_args["branch"]
            )
        except Exception as e:
            abort(400, message=str(e))
        return make_success_json_response()

    @write_route(f"{prefix}/agent_config/request_review", methods=["POST"])
    @enforce_policy(edit_policy)
    @blueprint.arguments(_RequestReviewPostJsonSchema, location="json")
    @blueprint.response(200)
    @obs.api_call()
    def _request_review(json_args: dict[str, str]) -> flask.Response:
        from components.ai_tooling.internal.agent_config_tool.business_logic.review_management import (
            request_review,
        )

        if not self._is_peer_review_enabled():
            abort(404)
        assert self._review_notifications_slack_channel is not None

        try:
            review_state = request_review(
                store=self._country_store(),
                env=json_args["env"],
                branch=json_args["branch"],
                commit_message=json_args["commit_message"],
                review_notifications_slack_channel=self._review_notifications_slack_channel,
                review_url=json_args["review_url"],
            )
        except Exception as e:
            abort(400, message=str(e))
        return make_json_response(review_state.to_dict())

    @blueprint.route(f"{prefix}/agent_config/review_state", methods=["GET"])
    @enforce_policy(view_policy)
    @blueprint.arguments(_GetReviewStateGetQuerySchema, location="query")
    @blueprint.response(200)
    @obs.api_call()
    def _get_review_state(query_args: dict[str, str]) -> flask.Response:
        from components.ai_tooling.internal.agent_config_tool.business_logic.review_management import (
            fetch_review_state,
        )

        is_enabled = self._is_peer_review_enabled()
        review_state = (
            fetch_review_state(
                store=self._country_store(),
                env=query_args.get("env") or get_env_name(),
                branch=query_args["branch"],
                version=query_args.get("version"),
            )
            if is_enabled
            else None
        )
        return make_json_response(
            {
                "is_peer_review_enabled": is_enabled,
                "review_state": review_state.to_dict() if review_state else None,
            }
        )

    @write_route(f"{prefix}/agent_config/approve_review", methods=["POST"])
    @enforce_policy(edit_policy)
    @blueprint.arguments(_ApproveReviewPostJsonSchema, location="json")
    @blueprint.response(200)
    @obs.api_call()
    def _approve_review(json_args: dict[str, Any]) -> flask.Response:
        from components.ai_tooling.internal.agent_config_tool.business_logic.review_management import (
            approve_review,
        )

        if not self._is_peer_review_enabled():
            abort(404)

        try:
            review_state = approve_review(
                store=self._country_store(),
                env=json_args["env"],
                branch=json_args["branch"],
                version=json_args["version"],
            )
        except Exception as e:
            abort(400, message=str(e))
        return make_json_response(review_state.to_dict())

    @blueprint.route(f"{prefix}/agent_config/agents", methods=["GET"])
    @enforce_policy(view_policy)
    @blueprint.arguments(_ListAgentsGetQuerySchema, location="query")
    @blueprint.response(200)
    @obs.api_call()
    def _list_agents(query_args: dict[str, str]) -> flask.Response:
        from components.ai_tooling.internal.agent_config_tool.business_logic.version_management import (
            resolve_env_branch_version,
        )

        try:
            store = self._country_store()
            env, branch, version = resolve_env_branch_version(
                store=store,
                env=query_args.get("env"),
                branch=query_args.get("branch"),
                version=query_args.get("version"),
            )
            names = store.list_agent_names(env, branch, version)
        except Exception as e:
            abort(400, message=str(e))
        return make_json_response(
            {"agents": [{"name": name} for name in names], "branch": branch}
        )

    @blueprint.route(f"{prefix}/agent_config/agent", methods=["GET"])
    @enforce_policy(view_policy)
    @blueprint.arguments(_GetAgentGetQuerySchema, location="query")
    @blueprint.response(200)
    @obs.api_call()
    def _get_agent(query_args: dict[str, str]) -> flask.Response:
        from components.ai_tooling.internal.agent_config_tool.business_logic.version_management import (
            resolve_env_branch_version,
        )

        try:
            store = self._country_store()
            env, branch, version = resolve_env_branch_version(
                store=store,
                env=query_args.get("env"),
                branch=query_args.get("branch"),
                version=query_args.get("version"),
            )
            configs = store.fetch_agent(query_args["name"], env, branch, version)
            for enrich in self._enrichers:
                configs = enrich(configs)
        except Exception as e:
            abort(400, message=str(e))
        return make_json_response(configs)

    @blueprint.route(f"{prefix}/agent_config/tools", methods=["GET"])
    @enforce_policy(view_policy)
    @blueprint.arguments(_GetAllGetQuerySchema, location="query")
    @blueprint.response(200)
    @obs.api_call()
    def _get_tools_catalog(query_args: dict[str, str]) -> flask.Response:
        from components.ai_tooling.internal.agent_config_tool.business_logic.version_management import (
            resolve_env_branch_version,
        )

        try:
            store = self._country_store()
            env, branch, version = resolve_env_branch_version(
                store=store,
                env=query_args.get("env"),
                branch=query_args.get("branch"),
                version=query_args.get("version"),
            )
            tools = store.fetch_tools_catalog(env, branch, version)
        except Exception as e:
            abort(400, message=str(e))
        return make_json_response({"tools": tools})

ConfigEnricher `module-attribute` ¶

ConfigEnricher = Callable[[dict[str, Any]], dict[str, Any]]

standard_sections_enricher ¶

standard_sections_enricher(get_agent_prompt_map)

Enricher expanding agents/member_attributes/redirections (S3-backed stores).

Source code in components/ai_tooling/public/ai_agent_config_tool.py

def standard_sections_enricher(
    get_agent_prompt_map: Callable[[], list[AgentPromptMapping]],
) -> ConfigEnricher:
    """Enricher expanding agents/member_attributes/redirections (S3-backed stores)."""

    def _enrich(configs: dict[str, Any]) -> dict[str, Any]:
        from components.ai_tooling.internal.agent_config_tool.helpers.config_enrichers import (
            enrich_standard_sections,
        )

        return enrich_standard_sections(configs, get_agent_prompt_map())

    return _enrich

components.ai_tooling.public.ai_debug_tool ¶

AIDebugTool ¶

AIDebugTool(
    get_doctorai_conversation_id,
    get_conversation_parts,
    get_context_sections,
    get_conversation_overview_config,
    paginate_conversations,
)

Configurable AI debug tool for conversations created using the DoctorAI service.

Provides a reusable framework for debugging AI conversations with a three-column layout: context panel, conversation view, and trace view. The trace is automatically pulled from DoctorAI's API.

Usage

debug_tool = AIDebugTool( get_doctorai_conversation_id=lambda entity_id: entity_id, get_conversation_elements=my_conversation_fn, get_context_sections=my_context_fn, ) debug_tool.register_smorest_routes(my_blueprint)

Initialize the AI debug tool.

Parameters:

Name	Type	Description	Default
`get_doctorai_conversation_id`	`Callable[[UUID], str \| None]`	Function to get DoctorAI conversation ID from entity ID.	required
`get_conversation_parts`	`Callable[[UUID], ConversationPanel]`	Function to get conversation elements for display.	required
`get_context_sections`	`Callable[[UUID], list[BaseContextSection]]`	Function to get context panel sections.	required
`get_conversation_overview_config`	`Callable[[], ConversationOverviewConfig]`	Function returning overview config (columns + card preview).	required
`paginate_conversations`	`Callable[[int, int, dict[str, str]], ConversationsOverviewResponse]`	Function returning paginated, filtered conversations.	required

Source code in components/ai_tooling/public/ai_debug_tool.py

def __init__(
    self,
    get_doctorai_conversation_id: Callable[[uuid.UUID], str | None],
    get_conversation_parts: Callable[[uuid.UUID], ConversationPanel],
    get_context_sections: Callable[[uuid.UUID], list[BaseContextSection]],
    get_conversation_overview_config: Callable[[], ConversationOverviewConfig],
    paginate_conversations: Callable[
        [int, int, dict[str, str]], ConversationsOverviewResponse
    ],
):
    """Initialize the AI debug tool.

    Args:
        get_doctorai_conversation_id: Function to get DoctorAI conversation ID from entity ID.
        get_conversation_parts: Function to get conversation elements for display.
        get_context_sections: Function to get context panel sections.
        get_conversation_overview_config: Function returning overview config (columns + card preview).
        paginate_conversations: Function returning paginated, filtered conversations.
    """
    self._get_doctorai_conversation_id = get_doctorai_conversation_id
    self._get_conversation_parts = get_conversation_parts
    self._get_context_sections = get_context_sections
    self._get_conversation_overview_config = get_conversation_overview_config
    self._paginate_conversations = paginate_conversations

register_smorest_routes ¶

register_smorest_routes(blueprint, prefix='')

Register debug tool routes on a flask-smorest blueprint.

Creates endpoints: - GET {prefix}/debug/{entity_id}/conversation - GET {prefix}/debug/{entity_id}/context - GET {prefix}/debug/{entity_id}/trace - GET {prefix}/debug/config - GET {prefix}/debug/conversations

Source code in components/ai_tooling/public/ai_debug_tool.py

def register_smorest_routes(
    self, blueprint: CustomBlueprint, prefix: str = ""
) -> None:
    """Register debug tool routes on a flask-smorest blueprint.

    Creates endpoints:
    - GET {prefix}/debug/{entity_id}/conversation
    - GET {prefix}/debug/{entity_id}/context
    - GET {prefix}/debug/{entity_id}/trace
    - GET {prefix}/debug/config
    - GET {prefix}/debug/conversations
    """
    prefix = prefix.rstrip("/")
    marmot_view_policy = BackofficePermissionAccessPolicy.permitted_for(
        {EmployeePermission.view_marmot_information}
    )

    def _route(rule_suffix: str) -> str:
        if prefix:
            return f"{prefix}{rule_suffix}"
        return rule_suffix

    @blueprint.route(
        _route("/debug/<strict_uuid:entity_id>/conversation"), methods=["GET"]
    )
    @enforce_policy(marmot_view_policy)
    @blueprint.response(200)
    @obs.api_call()
    def _get_debug_conversation(entity_id: uuid.UUID) -> Response:
        conversation = self._get_conversation_parts(entity_id)
        return make_json_response(conversation)

    @blueprint.route(
        _route("/debug/<strict_uuid:entity_id>/context"), methods=["GET"]
    )
    @enforce_policy(marmot_view_policy)
    @blueprint.response(200)
    @obs.api_call()
    def _get_debug_context(entity_id: uuid.UUID) -> Response:
        sections = self._get_context_sections(entity_id)
        return make_json_response(sections)

    @blueprint.route(
        _route("/debug/<strict_uuid:entity_id>/trace"), methods=["GET"]
    )
    @enforce_policy(marmot_view_policy)
    @blueprint.response(200)
    @blueprint.alt_response(404, description="Trace not found")
    @obs.api_call()
    def _get_debug_trace(entity_id: uuid.UUID) -> Response:
        doctorai_conversation_id = self._get_doctorai_conversation_id(entity_id)

        if not doctorai_conversation_id:
            abort(404)

        trace_data = get_agent_task_trees_in_conversation(doctorai_conversation_id)

        if trace_data is None:
            abort(404)

        return make_json_response(trace_data)

    @blueprint.route(_route("/debug/config"), methods=["GET"])
    @enforce_policy(marmot_view_policy)
    @blueprint.response(200)
    @obs.api_call()
    def _get_debug_config() -> Response:
        config = self._get_conversation_overview_config()
        return make_json_response(config)

    @blueprint.route(_route("/debug/conversations"), methods=["GET"])
    @enforce_policy(marmot_view_policy)
    @blueprint.arguments(DebugConversationsGetQuerySchema, location="query")
    @blueprint.response(200)
    @obs.api_call()
    def _get_debug_conversations(query_args: dict) -> Response:  # type: ignore[type-arg]
        filters: dict[str, str] = json.loads(str(query_args["filters"]))
        result = self._paginate_conversations(
            int(query_args["page"]),
            int(query_args["per_page"]),
            filters,
        )
        return make_json_response(result)

DebugConversationsGetQuerySchema ¶

Bases: Schema

Query parameters for the paginated conversations list endpoint.

filters `class-attribute` `instance-attribute` ¶

filters = Str(load_default='{}')

page `class-attribute` `instance-attribute` ¶

page = Int(load_default=1)

per_page `class-attribute` `instance-attribute` ¶

per_page = Int(load_default=50)

components.ai_tooling.public.ai_review_tool ¶

AIReviewTool ¶

AIReviewTool(
    agent_type,
    review_type,
    get_conversation_parts,
    get_context_sections,
    get_review_form,
    get_reviewable_conversations,
    get_reviewable_conversations_config,
    assign_conversation,
    get_and_assign_next_conversation,
    on_submit_review=None,
    required_permission=EmployeePermission.view_marmot_information,
    persist_review_response=True,
)

Configurable AI review tool for conversations created using the DoctorAI service.

Provides a reusable framework for reviewing AI conversations with a three-column layout: context panel, conversation view, and review form.

Usage

review_tool = AIReviewTool( get_conversation_parts=my_conversation_parts_fn, get_context_sections=my_context_fn, get_review_form=my_review_form_fn, get_reviewable_conversations=my_paginated_conversations_fn, on_submit_review=my_additional_submit_fn, # optional get_reviewable_conversations_config=my_config_fn, assign_conversation=my_assign_fn, ) review_tool.register_smorest_routes(my_controller)

Initialize the AI review tool.

Parameters:

Name	Type	Description	Default
`agent_type`	`AgentType`	The type of AI agent being reviewed.	required
`review_type`	`ReviewType`	The type of review workflow.	required
`get_conversation_parts`	`Callable[[UUID], ConversationPanel]`	Function to get conversation elements for display.	required
`get_context_sections`	`Callable[[UUID], list[BaseContextSection]]`	Function to get context panel sections.	required
`get_review_form`	`Callable[[UUID], ReviewForm]`	Function to get the review form configuration.	required
`on_submit_review`	`SubmitReviewCallable \| None`	Optional callback for additional submit logic. Called after the base submit_review_response. Receives (entity_id, reviewer_id, answers).	`None`
`get_reviewable_conversations`	`Callable[[int, int, dict[str, str], int \| None], ConversationsOverviewResponse]`	Function returning paginated, filtered conversations for review. Accepts (page, per_page, filters, user_id).	required
`get_reviewable_conversations_config`	`Callable[[], ConversationOverviewConfig]`	Function returning column configuration.	required
`assign_conversation`	`Callable[[UUID, int], AssignConversationResult]`	Function to assign a conversation to a reviewer (entity_id, user_id).	required
`get_and_assign_next_conversation`	`Callable[[dict[str, str], int], UUID \| None]`	Function to get and assign next unreviewed conversation. Accepts (filters, user_id) and returns entity_id or None if no more.	required
`required_permission`	`EmployeePermission`	Permission required to access the review tool endpoints.	`view_marmot_information`
`persist_review_response`	`bool`	Whether to store a generic AIReviewResponse row on submit. Defaults to True. Set False for tools whose on_submit_review already produces the durable record (e.g. annotation, which writes an eval TestCase).	`True`

Source code in components/ai_tooling/public/ai_review_tool.py

def __init__(
    self,
    agent_type: AgentType,
    review_type: ReviewType,
    get_conversation_parts: Callable[[uuid.UUID], ConversationPanel],
    get_context_sections: Callable[[uuid.UUID], list[BaseContextSection]],
    get_review_form: Callable[[uuid.UUID], ReviewForm],
    get_reviewable_conversations: Callable[
        [int, int, dict[str, str], int | None], ConversationsOverviewResponse
    ],
    get_reviewable_conversations_config: Callable[[], ConversationOverviewConfig],
    assign_conversation: Callable[[uuid.UUID, int], AssignConversationResult],
    get_and_assign_next_conversation: Callable[
        [dict[str, str], int], uuid.UUID | None
    ],
    on_submit_review: SubmitReviewCallable | None = None,
    required_permission: EmployeePermission = EmployeePermission.view_marmot_information,
    persist_review_response: bool = True,
):
    """Initialize the AI review tool.

    Args:
        agent_type: The type of AI agent being reviewed.
        review_type: The type of review workflow.
        get_conversation_parts: Function to get conversation elements for display.
        get_context_sections: Function to get context panel sections.
        get_review_form: Function to get the review form configuration.
        on_submit_review: Optional callback for additional submit logic. Called after
            the base submit_review_response. Receives (entity_id, reviewer_id, answers).
        get_reviewable_conversations: Function returning paginated, filtered conversations
            for review. Accepts (page, per_page, filters, user_id).
        get_reviewable_conversations_config: Function returning column configuration.
        assign_conversation: Function to assign a conversation to a reviewer (entity_id, user_id).
        get_and_assign_next_conversation: Function to get and assign next unreviewed conversation.
            Accepts (filters, user_id) and returns entity_id or None if no more.
        required_permission: Permission required to access the review tool endpoints.
        persist_review_response: Whether to store a generic AIReviewResponse row on submit.
            Defaults to True. Set False for tools whose on_submit_review already produces the
            durable record (e.g. annotation, which writes an eval TestCase).
    """
    self._review_type = review_type
    self._agent_type = agent_type
    self._get_conversation_parts = get_conversation_parts
    self._get_context_sections = get_context_sections
    self._get_review_form = get_review_form
    self._get_reviewable_conversations = get_reviewable_conversations
    self._on_submit_review = on_submit_review
    self._persist_review_response = persist_review_response
    self._get_reviewable_conversations_config = get_reviewable_conversations_config
    self._assign_conversation = assign_conversation
    self._get_and_assign_next_conversation = get_and_assign_next_conversation
    self._required_permission = required_permission

register_smorest_routes ¶

register_smorest_routes(blueprint, prefix='')

Register review tool routes on a flask-smorest blueprint.

Creates endpoints: - GET {prefix}/review/{entity_id}/conversation - GET {prefix}/review/{entity_id}/context - GET {prefix}/review/{entity_id}/form - POST {prefix}/review/{entity_id}/submit - POST {prefix}/review/{entity_id}/assign - POST {prefix}/review/{entity_id}/submit-and-next - GET {prefix}/review/config - GET {prefix}/review/conversations

Source code in components/ai_tooling/public/ai_review_tool.py

def register_smorest_routes(
    self, blueprint: CustomBlueprint, prefix: str = ""
) -> None:
    """Register review tool routes on a flask-smorest blueprint.

    Creates endpoints:
    - GET {prefix}/review/{entity_id}/conversation
    - GET {prefix}/review/{entity_id}/context
    - GET {prefix}/review/{entity_id}/form
    - POST {prefix}/review/{entity_id}/submit
    - POST {prefix}/review/{entity_id}/assign
    - POST {prefix}/review/{entity_id}/submit-and-next
    - GET {prefix}/review/config
    - GET {prefix}/review/conversations
    """
    prefix = prefix.rstrip("/")
    review_policy = BackofficePermissionAccessPolicy.permitted_for(
        {self._required_permission}
    )

    def _route(rule_suffix: str) -> str:
        if prefix:
            return f"{prefix}{rule_suffix}"
        return rule_suffix

    @blueprint.route(
        _route("/review/<uuid:entity_id>/conversation"), methods=["GET"]
    )
    @enforce_policy(review_policy)
    @blueprint.response(200)
    @obs.api_call()
    def _get_conversation_parts(entity_id: uuid.UUID) -> flask.Response:
        user_id: int | None = getattr(g, "actor", None) and g.actor.id
        assignment = self._assign_conversation(entity_id, mandatory(user_id))
        conversation = self._get_conversation_parts(entity_id)
        return make_json_response(
            {
                **conversation.to_dict(),
                "assignment": {
                    "status": assignment.status.value,
                    "assigned_to_user_id": assignment.assigned_to_user_id,
                },
            }
        )

    @blueprint.route(_route("/review/<uuid:entity_id>/context"), methods=["GET"])
    @enforce_policy(review_policy)
    @blueprint.response(200)
    @obs.api_call()
    def _get_review_context(entity_id: uuid.UUID) -> flask.Response:
        sections = self._get_context_sections(entity_id)
        return make_json_response(sections)

    @blueprint.route(_route("/review/<uuid:entity_id>/form"), methods=["GET"])
    @enforce_policy(review_policy)
    @blueprint.response(200)
    @obs.api_call()
    def _get_review_form(entity_id: uuid.UUID) -> flask.Response:
        form = self._get_review_form(entity_id)
        return make_json_response(form)

    @blueprint.route(_route("/review/<uuid:entity_id>/submit"), methods=["POST"])
    @enforce_policy(review_policy)
    @blueprint.arguments(SubmitReviewPostJsonSchema, location="json")
    @blueprint.response(200)
    @obs.api_call()
    def _submit_review(
        json_args: dict[str, Any], entity_id: uuid.UUID
    ) -> flask.Response:
        reviewer_id = str(g.current_user.id)
        answers = json_args["answers"]

        result = self._submit(entity_id, reviewer_id, answers)

        return make_json_response(result)

    @blueprint.route(_route("/review/<uuid:entity_id>/assign"), methods=["POST"])
    @enforce_policy(review_policy)
    @blueprint.response(200)
    @obs.api_call()
    def _assign_conversation(entity_id: uuid.UUID) -> flask.Response:
        user_id: int | None = getattr(g, "actor", None) and g.actor.id
        result = self._assign_conversation(entity_id, mandatory(user_id))
        response_data = {
            "status": result.status.value,
            "assigned_to_user_id": result.assigned_to_user_id,
        }
        return make_json_response(response_data)

    @blueprint.route(
        _route("/review/<uuid:entity_id>/submit-and-next"), methods=["POST"]
    )
    @enforce_policy(review_policy)
    @blueprint.arguments(SubmitAndNextPostJsonSchema, location="json")
    @blueprint.response(200)
    @obs.api_call()
    def _submit_and_next(
        json_args: dict[str, Any], entity_id: uuid.UUID
    ) -> flask.Response:
        reviewer_id = str(g.current_user.id)
        user_id: int = mandatory(getattr(g, "actor", None) and g.actor.id)
        answers = json_args["answers"]
        filters: dict[str, str] = json.loads(str(json_args["filters"]))

        result = self._submit(entity_id, reviewer_id, answers)

        next_entity_id = self._get_and_assign_next_conversation(filters, user_id)

        return make_json_response(
            SubmitAndNextResult(
                review_id=result.id if result else None,
                next_entity_id=next_entity_id,
            )
        )

    @blueprint.route(_route("/review/config"), methods=["GET"])
    @enforce_policy(review_policy)
    @blueprint.response(200)
    @obs.api_call()
    def _get_reviewable_conversations_config() -> flask.Response:
        config = self._get_reviewable_conversations_config()
        return make_json_response(config)

    @blueprint.route(_route("/review/conversations"), methods=["GET"])
    @enforce_policy(review_policy)
    @blueprint.arguments(ReviewConversationsGetQuerySchema, location="query")
    @blueprint.response(200)
    @obs.api_call()
    def _get_reviewable_conversations(query_args: dict[str, Any]) -> flask.Response:
        filters: dict[str, str] = json.loads(str(query_args["filters"]))
        user_id: int | None = getattr(g, "actor", None) and g.actor.id
        result = self._get_reviewable_conversations(
            int(query_args["page"]),
            int(query_args["per_page"]),
            filters,
            user_id,
        )
        return make_json_response(result)

AssignConversationResult `dataclass` ¶

AssignConversationResult(status, assigned_to_user_id=None)

Result of attempting to assign a conversation for review.

assigned_to_user_id `class-attribute` `instance-attribute` ¶

assigned_to_user_id = None

status `instance-attribute` ¶

status

AssignmentStatus ¶

Bases: Enum

Status of a conversation assignment attempt.

ALREADY_ASSIGNED_TO_YOU `class-attribute` `instance-attribute` ¶

ALREADY_ASSIGNED_TO_YOU = 'already_assigned_to_you'

ASSIGNED `class-attribute` `instance-attribute` ¶

ASSIGNED = 'assigned'

CONFLICT `class-attribute` `instance-attribute` ¶

CONFLICT = 'conflict'

ReviewConversationsGetQuerySchema ¶

Bases: Schema

Query parameters for the paginated conversations list endpoint.

filters `class-attribute` `instance-attribute` ¶

filters = Str(load_default='{}')

page `class-attribute` `instance-attribute` ¶

page = Int(load_default=1)

per_page `class-attribute` `instance-attribute` ¶

per_page = Int(load_default=50)

SubmitAndNextPostJsonSchema ¶

Bases: Schema

JSON body for submitting a review and getting the next conversation.

answers `class-attribute` `instance-attribute` ¶

answers = Dict(keys=Str(), required=True)

filters `class-attribute` `instance-attribute` ¶

filters = Str(load_default='{}')

SubmitAndNextResult `dataclass` ¶

SubmitAndNextResult(review_id=None, next_entity_id=None)

Bases: DataClassJsonMixin

Result of submitting a review and getting the next conversation.

next_entity_id `class-attribute` `instance-attribute` ¶

next_entity_id = None

review_id `class-attribute` `instance-attribute` ¶

review_id = None

SubmitReviewCallable `module-attribute` ¶

SubmitReviewCallable = Callable[
    [UUID, str, dict[str, Any]], None
]

SubmitReviewPostJsonSchema ¶

Bases: Schema

JSON body for submitting a review response.

answers `class-attribute` `instance-attribute` ¶

answers = Dict(keys=Str(), required=True)

SubmitReviewResult `dataclass` ¶

SubmitReviewResult(id, entity_id)

Result of submitting a review response.

entity_id `instance-attribute` ¶

entity_id

id `instance-attribute` ¶

id

components.ai_tooling.public.ai_tooling_app_group ¶

ai_tooling `module-attribute` ¶

ai_tooling = AppGroup('ai_tooling')

components.ai_tooling.public.ai_tooling_service ¶

Service encapsulating dedicated-store prompt fetching.

Background: agent prompts historically lived in a single automated-resolution/fr/... S3 prefix (the "AR store") and were fetched by FR's legacy get_llm_prompt_template. We're migrating one agent at a time to dedicated stores (legal-complaint/fr/..., future harry/fr/..., ...). This service owns the "which prompts are served by dedicated stores, and how to fetch them" routing, keeping that logic out of FR-specific files.

FR's get_llm_prompt_template is expected to delegate to this service when is_prompt_template_supported returns True, otherwise fall through to its existing AR-store code path.

AiToolingService ¶

Encapsulates dedicated-store prompt fetching, isolated from FR's legacy AR-store code path.

get_llm_prompt_template `staticmethod` ¶

get_llm_prompt_template(
    prompt_name,
    config_branch=None,
    config_version=None,
    config_env=None,
)

Return the raw prompt template dict from the prompt's dedicated store.

Caller is responsible for converting to its own entity type (e.g. LlmPromptTemplate.from_dict(...) on the FR side). Raises if the prompt is not routed to a dedicated store — callers should guard with is_prompt_template_supported first.

Source code in components/ai_tooling/public/ai_tooling_service.py

@staticmethod
def get_llm_prompt_template(
    prompt_name: str,
    config_branch: str | None = None,
    config_version: str | None = None,
    config_env: str | None = None,
) -> dict[str, Any]:
    """Return the raw prompt template dict from the prompt's dedicated store.

    Caller is responsible for converting to its own entity type
    (e.g. `LlmPromptTemplate.from_dict(...)` on the FR side). Raises if the
    prompt is not routed to a dedicated store — callers should guard with
    `is_prompt_template_supported` first.
    """
    agent_store = _get_store_for_prompt(prompt_name)
    if agent_store is None:
        raise ValueError(
            f"Prompt '{prompt_name}' is not served by a dedicated store. "
            f"Call `is_prompt_template_supported` first."
        )
    store = agent_store.for_country(app_name=AppName.ALAN_FR)
    env, branch, version = get_nonnull_env_branch_version(
        env=config_env, branch=config_branch, version=config_version, store=store
    )
    templates_data = store.fetch_section(
        ConfigSection.llm_prompt_templates.value, env, branch, version
    )
    template_data: dict[str, Any] | None = next(
        (t for t in templates_data if t.get("name") == prompt_name), None
    )
    if template_data is None:
        raise Exception(
            f"LLM prompt template '{prompt_name}' not found in store "
            f"prefix='{agent_store.prefix}' branch={branch} version={version}"
        )
    return template_data

is_prompt_template_supported `staticmethod` ¶

is_prompt_template_supported(prompt_name)

True iff this prompt is served by a dedicated store (not the AR fallback).

Source code in components/ai_tooling/public/ai_tooling_service.py

@staticmethod
def is_prompt_template_supported(prompt_name: str) -> bool:
    """True iff this prompt is served by a dedicated store (not the AR fallback)."""
    return _get_agent_name_from_prompt_name(prompt_name) is not None

resolve_env_branch_version_for_prompt `staticmethod` ¶

resolve_env_branch_version_for_prompt(
    prompt_name, env=None, branch=None, version=None
)

Resolve (env, branch, version) against the prompt's dedicated store.

Used by upstream callers that pin a version before calling doctorai so the round-trip stays consistent. Raises if the prompt is not routed to a dedicated store.

Source code in components/ai_tooling/public/ai_tooling_service.py

@staticmethod
def resolve_env_branch_version_for_prompt(
    prompt_name: str,
    env: str | None = None,
    branch: str | None = None,
    version: str | None = None,
) -> tuple[str, str, str]:
    """Resolve (env, branch, version) against the prompt's dedicated store.

    Used by upstream callers that pin a version before calling doctorai so
    the round-trip stays consistent. Raises if the prompt is not routed to
    a dedicated store.
    """
    agent_store = _get_store_for_prompt(prompt_name)
    if agent_store is None:
        raise ValueError(
            f"Prompt '{prompt_name}' is not served by a dedicated store."
        )
    store = agent_store.for_country(app_name=AppName.ALAN_FR)
    return get_nonnull_env_branch_version(
        env=env, branch=branch, version=version, store=store
    )

components.ai_tooling.public.blueprint ¶

ai_tooling_blueprint `module-attribute` ¶

ai_tooling_blueprint = create_blueprint(
    name="ai_tooling",
    import_name=__name__,
    template_folder=join(
        dirname(__file__), "..", "templates"
    ),
    cli_group="ai_tooling",
)

components.ai_tooling.public.commands ¶

seed_agent_config_store ¶

Seed an empty initial commit in an agent config S3 store.

Required once per agent config store before the Marmot Agents Config tab can load — resolve_env_branch_version raises "No versions found" on an empty store. Seeds dev, stage and prod in one go (bucket is shared across envs; the env is just a path segment).

Auto-scopes to the current app's country prefix (matches the runtime behavior of AIAgentConfigTool._country_store).

Usage

flask ai_tooling seed-agent-config-store --store legal-complaint flask ai_tooling seed-agent-config-store --store harry --country fr

SEEDED_ENVS `module-attribute` ¶

SEEDED_ENVS = ('dev', 'stage', 'prod')

seed_agent_config_store ¶

seed_agent_config_store(
    dry_run=False, store_prefix="", country_value=None
)

Seed {store_prefix}/{country}/{env}/main/ with an empty initial commit for each of dev, stage, prod.

No-ops on any env whose branch already has a commit history.

Source code in components/ai_tooling/public/commands/seed_agent_config_store.py

@ai_tooling.command()
@click.option(
    "--store",
    "store_prefix",
    required=True,
    help="S3 prefix of the agent config store to seed (e.g. 'legal-complaint').",
)
@click.option(
    "--country",
    "country_value",
    type=str,
    default=None,
    help="Country sub-prefix (e.g. 'fr', 'be'). Defaults to the current app's country.",
)
@command_with_dry_run
def seed_agent_config_store(
    dry_run: bool = False,
    store_prefix: str = "",
    country_value: str | None = None,
) -> None:
    """Seed `{store_prefix}/{country}/{env}/main/` with an empty initial commit for each of dev, stage, prod.

    No-ops on any env whose branch already has a commit history.
    """
    branch: str = "main"
    stores_by_prefix = _known_stores()
    if store_prefix not in stores_by_prefix:
        raise click.BadParameter(
            f"Unknown store '{store_prefix}'. Known stores: {sorted(stores_by_prefix)}"
        )

    if country_value is None:
        country = Country.from_app_name(get_current_app_name())
    else:
        try:
            country = Country(country_value)
        except ValueError as e:
            raise click.BadParameter(
                f"Unknown country '{country_value}'. Known: {[c.value for c in Country]}"
            ) from e

    store = stores_by_prefix[store_prefix].with_sub_prefix(country.value)

    for env in SEEDED_ENVS:
        existing_history = store.fetch_commit_history(env, branch)
        if existing_history:
            current_logger.info(
                "seed_agent_config_store: branch already initialized, no-op",
                store=store_prefix,
                country=country.value,
                env=env,
                branch=branch,
                latest_version=existing_history[-1],
            )
            continue

        metadata = AgentConfigCommitMetadata.create(f"seed {store_prefix} store")
        sections_data: dict[ConfigSection, list[Any]] = {
            section: [] for section in store.sections
        }

        if dry_run:
            current_logger.info(
                "seed_agent_config_store: would write initial commit",
                store=store_prefix,
                country=country.value,
                env=env,
                branch=branch,
                version=metadata.created_at,
                sections=[s.value for s in store.sections],
                dry_run=True,
            )
            continue

        store.upload_all_sections(
            sections_data, metadata, env=env, branch=branch, version=metadata.created_at
        )
        current_logger.info(
            "seed_agent_config_store: wrote initial commit",
            store=store_prefix,
            country=country.value,
            env=env,
            branch=branch,
            version=metadata.created_at,
        )

components.ai_tooling.public.config_store ¶

ConfigStore ¶

Bases: Protocol

Backend-agnostic agent-config store contract used by AIAgentConfigTool.

copy ¶

copy(
    source_env,
    source_branch,
    target_env,
    target_branch,
    metadata,
    source_version=None,
)

Copy a source's configs to a target as a new commit.

Source code in components/ai_tooling/public/config_store.py

def copy(
    self,
    source_env: str | None,
    source_branch: str,
    target_env: str | None,
    target_branch: str,
    metadata: AgentConfigCommitMetadata,
    source_version: str | None = None,
) -> None:
    """Copy a source's configs to a target as a new commit."""
    ...

delete_branch ¶

delete_branch(env, branch)

Delete all stored data for a branch.

Source code in components/ai_tooling/public/config_store.py

def delete_branch(self, env: str, branch: str) -> None:
    """Delete all stored data for a branch."""
    ...

fetch_agent ¶

fetch_agent(name, env, branch, version)

Fetch a single agent's sections + commit metadata for a version.

Source code in components/ai_tooling/public/config_store.py

def fetch_agent(
    self, name: str, env: str, branch: str, version: str
) -> dict[str, Any]:
    """Fetch a single agent's sections + commit metadata for a version."""
    ...

fetch_all ¶

fetch_all(env, branch, version)

Fetch every section plus commit metadata for a version.

Source code in components/ai_tooling/public/config_store.py

def fetch_all(self, env: str, branch: str, version: str) -> dict[str, Any]:
    """Fetch every section plus commit metadata for a version."""
    ...

fetch_recent_commits ¶

fetch_recent_commits(env, branch, version, limit=10)

Fetch the most recent commits up to a version, newest first.

Source code in components/ai_tooling/public/config_store.py

def fetch_recent_commits(
    self, env: str, branch: str, version: str, limit: int = 10
) -> list[AgentConfigCommitMetadata]:
    """Fetch the most recent commits up to a version, newest first."""
    ...

fetch_review_state ¶

fetch_review_state(env, branch, version)

Fetch the review state for a version, or None if absent.

Source code in components/ai_tooling/public/config_store.py

def fetch_review_state(
    self, env: str, branch: str, version: str
) -> ReviewState | None:
    """Fetch the review state for a version, or None if absent."""
    ...

fetch_section ¶

fetch_section(section, env, branch, version)

Fetch a single section's data for a version.

Source code in components/ai_tooling/public/config_store.py

def fetch_section(self, section: str, env: str, branch: str, version: str) -> Any:
    """Fetch a single section's data for a version."""
    ...

fetch_tools_catalog ¶

fetch_tools_catalog(env, branch, version)

Fetch the full tool catalog for a version.

Source code in components/ai_tooling/public/config_store.py

def fetch_tools_catalog(
    self, env: str, branch: str, version: str
) -> list[dict[str, Any]]:
    """Fetch the full tool catalog for a version."""
    ...

for_country ¶

for_country(app_name=None)

Return a store scoped to a country (the current request's by default).

Source code in components/ai_tooling/public/config_store.py

def for_country(self, app_name: AppName | None = None) -> "ConfigStore":
    """Return a store scoped to a country (the current request's by default)."""
    ...

get_latest_version ¶

get_latest_version(env, branch)

Return a branch's latest version, or None if it has none.

Source code in components/ai_tooling/public/config_store.py

def get_latest_version(self, env: str, branch: str) -> str | None:
    """Return a branch's latest version, or None if it has none."""
    ...

list_agent_names ¶

list_agent_names(env, branch, version)

List the agent names for a version.

Source code in components/ai_tooling/public/config_store.py

def list_agent_names(self, env: str, branch: str, version: str) -> list[str]:
    """List the agent names for a version."""
    ...

list_branches ¶

list_branches(env)

List the branches available for an environment.

Source code in components/ai_tooling/public/config_store.py

def list_branches(self, env: str) -> list[str]:
    """List the branches available for an environment."""
    ...

remove_version ¶

remove_version(env, branch, version)

Remove a single version's data and history entry.

Source code in components/ai_tooling/public/config_store.py

def remove_version(self, env: str, branch: str, version: str) -> None:
    """Remove a single version's data and history entry."""
    ...

rollback ¶

rollback(target_version, env, branch, metadata)

Re-publish an older version as a new commit.

Source code in components/ai_tooling/public/config_store.py

def rollback(
    self,
    target_version: str,
    env: str,
    branch: str,
    metadata: AgentConfigCommitMetadata,
) -> None:
    """Re-publish an older version as a new commit."""
    ...

sections `instance-attribute` ¶

sections

upload_all_sections ¶

upload_all_sections(
    sections_data, metadata, env, branch, version
)

Publish a commit: all sections plus its metadata.

Source code in components/ai_tooling/public/config_store.py

def upload_all_sections(
    self,
    sections_data: dict[ConfigSection, Any],
    metadata: AgentConfigCommitMetadata,
    env: str,
    branch: str,
    version: str,
) -> None:
    """Publish a commit: all sections plus its metadata."""
    ...

upload_review_state ¶

upload_review_state(data, env, branch, version)

Persist the review state for a version.

Source code in components/ai_tooling/public/config_store.py

def upload_review_state(
    self, data: ReviewState, env: str, branch: str, version: str
) -> None:
    """Persist the review state for a version."""
    ...

components.ai_tooling.public.constants ¶

REQUIRE_PEER_REVIEW_FEATURE_FLAG `module-attribute` ¶

REQUIRE_PEER_REVIEW_FEATURE_FLAG = (
    "agent-config-require-peer-review"
)

components.ai_tooling.public.dependencies ¶

AIToolingDependency ¶

Bases: ABC

Country-specific data needed by platform-level AI tooling.

get_default_redirection_fields `abstractmethod` ¶

get_default_redirection_fields()

Return default field values for new redirection entries.

Source code in components/ai_tooling/public/dependencies.py

@abstractmethod
def get_default_redirection_fields(self) -> dict[str, str]:
    """Return default field values for new redirection entries."""

get_redirection_screen_names `abstractmethod` ¶

get_redirection_screen_names()

Return all redirection screen name enum values.

Source code in components/ai_tooling/public/dependencies.py

@abstractmethod
def get_redirection_screen_names(self) -> list[str]:
    """Return all redirection screen name enum values."""

COMPONENT_NAME `module-attribute` ¶

COMPONENT_NAME = 'ai_tooling'

get_app_dependency ¶

get_app_dependency()

Source code in components/ai_tooling/public/dependencies.py

def get_app_dependency() -> AIToolingDependency:  # noqa: D103
    return cast("CustomFlask", current_app).get_component_dependency(COMPONENT_NAME)  # type: ignore[no-any-return]

set_app_dependency ¶

set_app_dependency(dependency)

Source code in components/ai_tooling/public/dependencies.py

def set_app_dependency(dependency: AIToolingDependency) -> None:  # noqa: D103
    cast("CustomFlask", current_app).add_component_dependency(
        COMPONENT_NAME, dependency
    )

components.ai_tooling.public.entities ¶

agent_config ¶

AgentConfigCommitMetadata `dataclass` ¶

AgentConfigCommitMetadata(
    author,
    created_at,
    message,
    review_skipped=False,
    env=None,
    branch=None,
    version=None,
)

Bases: DataClassJsonMixin

Metadata describing a versioned commit, stored alongside config sections in S3.

author `instance-attribute` ¶

author

branch `class-attribute` `instance-attribute` ¶

branch = None

create `classmethod` ¶

create(message=None, review_skipped=False)

Build new commit metadata, stamped with the current user and time.

Source code in components/ai_tooling/public/entities/agent_config.py

@classmethod
def create(
    cls,
    message: str | None = None,
    review_skipped: bool = False,
) -> "AgentConfigCommitMetadata":
    """Build new commit metadata, stamped with the current user and time."""
    from flask import g

    from shared.helpers.time.utc import utcnow

    return cls(
        author=getattr(g, "alan_employee_email", "") or "",
        created_at=utcnow().strftime("%Y-%m-%dT%H:%M:%S"),
        message=message or "",
        review_skipped=review_skipped,
    )

created_at `instance-attribute` ¶

created_at

env `class-attribute` `instance-attribute` ¶

env = None

message `instance-attribute` ¶

message

review_skipped `class-attribute` `instance-attribute` ¶

review_skipped = False

version `class-attribute` `instance-attribute` ¶

version = None

LlmParams `dataclass` ¶

LlmParams(
    provider_and_model, temperature, reasoning_effort=None
)

Bases: DataClassJsonMixin

LLM parameters for a prompt template.

provider_and_model `instance-attribute` ¶

provider_and_model

reasoning_effort `class-attribute` `instance-attribute` ¶

reasoning_effort = None

temperature `instance-attribute` ¶

temperature

PromptTemplate `dataclass` ¶

PromptTemplate(id, name, version, prompt_template, params)

Bases: DataClassJsonMixin

Named, versioned prompt template.

id `instance-attribute` ¶

id

name `instance-attribute` ¶

name

params `instance-attribute` ¶

params

prompt_template `instance-attribute` ¶

prompt_template

version `instance-attribute` ¶

version

ReviewState `dataclass` ¶

ReviewState(
    status,
    requester,
    commit_message,
    version,
    reviewer=None,
)

Bases: DataClassJsonMixin

Review state stored alongside config versions in S3.

commit_message `instance-attribute` ¶

commit_message

requester `instance-attribute` ¶

requester

reviewer `class-attribute` `instance-attribute` ¶

reviewer = None

status `instance-attribute` ¶

status

version `instance-attribute` ¶

version

ReviewStatus ¶

Bases: AlanBaseEnum

Status of a ReviewState for an agent-config version.

approved `class-attribute` `instance-attribute` ¶

approved = 'approved'

pending `class-attribute` `instance-attribute` ¶

pending = 'pending'

agent_prompt_mapping ¶

AgentPromptMapping `dataclass` ¶

AgentPromptMapping(
    agent_name, prompt_template_names=list()
)

Bases: DataClassJsonMixin

Agent name with its associated prompt template names.

agent_name `instance-attribute` ¶

agent_name

prompt_template_names `class-attribute` `instance-attribute` ¶

prompt_template_names = field(default_factory=list)

context_sections ¶

BaseContextSection `dataclass` ¶

BaseContextSection(*, title, section_type)

Bases: DataClassJsonMixin

Represents a section in the context panel.

Parameters:

Name	Type	Description	Default
`title`	`str`	Section title displayed in the UI.	required

section_type `instance-attribute` ¶

section_type

title `instance-attribute` ¶

title

BaseContextSectionType ¶

Bases: AlanBaseEnum

Types of elements that can appear as a section in the context panel

ContextSectionType ¶

Bases: BaseContextSectionType

Concrete context section types for the debug tool UI.

INFORMATION_LIST `class-attribute` `instance-attribute` ¶

INFORMATION_LIST = 'information_list'

ListContextSection `dataclass` ¶

ListContextSection(*, title, section_type, list_items)

Bases: BaseContextSection

A context section containing a list of items.

list_items `instance-attribute` ¶

list_items

ListItem `dataclass` ¶

ListItem(
    *, text, link_url=None, copy_value=None, copy_label=None
)

Bases: DataClassJsonMixin

A single item in a list context section.

copy_label `class-attribute` `instance-attribute` ¶

copy_label = None

copy_value `class-attribute` `instance-attribute` ¶

copy_value = None

link_url `class-attribute` `instance-attribute` ¶

link_url = None

text `instance-attribute` ¶

text

conversation_parts ¶

AgentHandoverConversationPart `dataclass` ¶

AgentHandoverConversationPart(
    *,
    id,
    part_type,
    timestamp,
    agent_task_id=None,
    from_agent,
    to_agent
)

Bases: BaseConversationPart

A conversation part representing a handover between agents.

from_agent `instance-attribute` ¶

from_agent

part_type `class-attribute` `instance-attribute` ¶

part_type = AGENT_HANDOVER

to_agent `instance-attribute` ¶

to_agent

BaseConversationPart `dataclass` ¶

BaseConversationPart(
    *, id, part_type, timestamp, agent_task_id=None
)

Bases: DataClassJsonMixin

Represents a single element in the conversation view.

Parameters:

Name	Type	Description	Default
`part_type`	`BaseConversationPartType`	The type of conversation element.	required
`timestamp`	`datetime \| None`	When this element occurred.	required

agent_task_id `class-attribute` `instance-attribute` ¶

agent_task_id = None

id `instance-attribute` ¶

id

part_type `instance-attribute` ¶

part_type

timestamp `instance-attribute` ¶

timestamp

BaseConversationPartType ¶

Bases: AlanBaseEnum

Types of elements that can appear in a conversation view.

ConversationPanel `dataclass` ¶

ConversationPanel(
    conversation, roles=None, specialized_engine=None
)

Bases: DataClassJsonMixin

Container for the full conversation view with optional role labels.

conversation `instance-attribute` ¶

conversation

roles `class-attribute` `instance-attribute` ¶

roles = None

specialized_engine `class-attribute` `instance-attribute` ¶

specialized_engine = None

ConversationPartType ¶

Bases: BaseConversationPartType

Concrete conversation part types for the debug tool UI.

AGENT_HANDOVER `class-attribute` `instance-attribute` ¶

AGENT_HANDOVER = 'agent_handover'

DOCUMENTS `class-attribute` `instance-attribute` ¶

DOCUMENTS = 'documents'

INFO_CARD `class-attribute` `instance-attribute` ¶

INFO_CARD = 'info_card'

MESSAGE `class-attribute` `instance-attribute` ¶

MESSAGE = 'message'

REASONING `class-attribute` `instance-attribute` ¶

REASONING = 'reasoning'

TOOL_CALL `class-attribute` `instance-attribute` ¶

TOOL_CALL = 'tool_call'

DocumentsConversationPart `dataclass` ¶

DocumentsConversationPart(
    *,
    id,
    part_type,
    timestamp,
    agent_task_id=None,
    role,
    documents
)

Bases: BaseConversationPart

A conversation part containing document attachments.

documents `instance-attribute` ¶

documents

part_type `class-attribute` `instance-attribute` ¶

part_type = DOCUMENTS

role `instance-attribute` ¶

role

InfoCardConversationPart `dataclass` ¶

InfoCardConversationPart(
    *,
    id,
    part_type,
    timestamp,
    agent_task_id=None,
    role,
    title,
    subtitle,
    icon=None,
    url_path=None
)

Bases: BaseConversationPart

A conversation part displaying an info card in the debug tool UI.

icon `class-attribute` `instance-attribute` ¶

icon = None

part_type `class-attribute` `instance-attribute` ¶

part_type = INFO_CARD

role `instance-attribute` ¶

role

subtitle `instance-attribute` ¶

subtitle

title `instance-attribute` ¶

title

url_path `class-attribute` `instance-attribute` ¶

url_path = None

MessageAttachment `dataclass` ¶

MessageAttachment(id, uri, type, name)

A file or document attached to a message.

id `instance-attribute` ¶

id

name `instance-attribute` ¶

name

type `instance-attribute` ¶

type

uri `instance-attribute` ¶

uri

MessageConversationPart `dataclass` ¶

MessageConversationPart(
    *,
    id,
    part_type,
    timestamp,
    agent_task_id=None,
    role,
    message,
    is_voice_transcription=None
)

Bases: BaseConversationPart

A text message in the conversation.

is_voice_transcription `class-attribute` `instance-attribute` ¶

is_voice_transcription = None

message `instance-attribute` ¶

message

part_type `class-attribute` `instance-attribute` ¶

part_type = MESSAGE

role `instance-attribute` ¶

role

MessageRole ¶

Bases: AlanBaseEnum

Role of a message sender in the conversation.

ai `class-attribute` `instance-attribute` ¶

ai = 'ai'

care_agent `class-attribute` `instance-attribute` ¶

care_agent = 'care_agent'

member `class-attribute` `instance-attribute` ¶

member = 'member'

ReasoningConversationPart `dataclass` ¶

ReasoningConversationPart(
    *,
    id,
    part_type,
    timestamp,
    agent_task_id=None,
    reasoning,
    role
)

Bases: BaseConversationPart

A conversation part representing LLM reasoning.

part_type `class-attribute` `instance-attribute` ¶

part_type = REASONING

reasoning `instance-attribute` ¶

reasoning

role `instance-attribute` ¶

role

ToolCallConversationPart `dataclass` ¶

ToolCallConversationPart(
    *,
    id,
    part_type,
    timestamp,
    agent_task_id=None,
    name,
    arguments,
    result,
    role
)

Bases: BaseConversationPart

A conversation part representing an LLM tool call and its result.

arguments `instance-attribute` ¶

arguments

name `instance-attribute` ¶

name

part_type `class-attribute` `instance-attribute` ¶

part_type = TOOL_CALL

result `instance-attribute` ¶

result

role `instance-attribute` ¶

role

ToolCallReviewStatus ¶

Bases: AlanBaseEnum

Human review state for a tool call with human review (metadata review_status).

Values must stay aligned with doctorai.common.enums.tool_call_review_status.

approved `class-attribute` `instance-attribute` ¶

approved = 'approved'

denied `class-attribute` `instance-attribute` ¶

denied = 'denied'

pending_review `class-attribute` `instance-attribute` ¶

pending_review = 'pending_review'

ToolCallWithReviewConversationPart `dataclass` ¶

ToolCallWithReviewConversationPart(
    *,
    id,
    part_type=ConversationPartType.TOOL_CALL,
    timestamp,
    agent_task_id=None,
    name,
    arguments,
    result,
    role,
    review_status=None,
    is_success=None
)

Bases: ToolCallConversationPart

A tool call with optional human review status and success flag.

is_success `class-attribute` `instance-attribute` ¶

is_success = None

part_type `class-attribute` `instance-attribute` ¶

part_type = TOOL_CALL

review_status `class-attribute` `instance-attribute` ¶

review_status = None

overview ¶

CardPreviewConfig `dataclass` ¶

CardPreviewConfig(title, timestamp, subtitle)

Bases: DataClassJsonMixin

Column names used for the conversation list card preview.

Parameters:

Name	Type	Description	Default
`title`	`str`	Column name for the top-left bold text.	required
`timestamp`	`str`	Column name for the top-right dimmed time.	required
`subtitle`	`str`	Column name for the truncated line below.	required

subtitle `instance-attribute` ¶

subtitle

timestamp `instance-attribute` ¶

timestamp

title `instance-attribute` ¶

title

ColumnConfig `dataclass` ¶

ColumnConfig(
    name,
    label,
    column_type,
    show_in_table,
    show_in_filter=True,
    options=None,
    link_template=None,
    required=False,
    default_value=None,
)

Bases: DataClassJsonMixin

Configuration for a single column in the conversation overview table.

Parameters:

Name	Type	Description	Default
`name`	`str`	Internal column identifier (e.g. "state").	required
`label`	`str`	Human-readable label (e.g. "State").	required
`column_type`	`ColumnType`	Data type of the column.	required
`show_in_table`	`bool`	Whether to display as a table column.	required
`show_in_filter`	`bool`	Whether to display as a filter (defaults to True for filterable types).	`True`
`options`	`list[str] \| None`	Enum values for enum-type columns, None otherwise.	`None`
`link_template`	`str \| None`	URL template with {value} placeholder for link columns.	`None`
`required`	`bool`	If True, filter cannot be cleared and default_value is applied.	`False`
`default_value`	`str \| None`	Default filter value for required columns.	`None`

column_type `instance-attribute` ¶

column_type

default_value `class-attribute` `instance-attribute` ¶

default_value = None

label `instance-attribute` ¶

label

link_template `class-attribute` `instance-attribute` ¶

link_template = None

name `instance-attribute` ¶

name

options `class-attribute` `instance-attribute` ¶

options = None

required `class-attribute` `instance-attribute` ¶

required = False

show_in_filter `class-attribute` `instance-attribute` ¶

show_in_filter = True

show_in_table `instance-attribute` ¶

show_in_table

ColumnType ¶

Bases: AlanBaseEnum

Data type for a column in the conversation overview table.

bool `class-attribute` `instance-attribute` ¶

bool = 'bool'

datetime `class-attribute` `instance-attribute` ¶

datetime = 'datetime'

enum `class-attribute` `instance-attribute` ¶

enum = 'enum'

search `class-attribute` `instance-attribute` ¶

search = 'search'

str `class-attribute` `instance-attribute` ¶

str = 'str'

ConversationListItem `dataclass` ¶

ConversationListItem(id, values)

Bases: DataClassJsonMixin

A single conversation row in the overview list.

Parameters:

Name	Type	Description	Default
`id`	`str`	Conversation identifier.	required
`values`	`dict[str, str \| None]`	Mapping of column_name to stringified value.	required

id `instance-attribute` ¶

id

values `instance-attribute` ¶

values

ConversationOverviewConfig `dataclass` ¶

ConversationOverviewConfig(columns, card_preview=None)

Bases: DataClassJsonMixin

Full configuration returned by get_conversation_overview_config.

Parameters:

Name	Type	Description	Default
`columns`	`list[ColumnConfig]`	Column definitions for the overview table.	required
`card_preview`	`CardPreviewConfig \| None`	Column mapping for the conversation list card preview.	`None`

card_preview `class-attribute` `instance-attribute` ¶

card_preview = None

columns `instance-attribute` ¶

columns

ConversationsOverviewResponse `dataclass` ¶

ConversationsOverviewResponse(items, page, per_page, total)

Bases: DataClassJsonMixin

Paginated response for the conversation overview endpoint.

Parameters:

Name	Type	Description	Default
`items`	`list[ConversationListItem]`	List of conversation rows for the current page.	required
`page`	`int`	Current page number (1-indexed).	required
`per_page`	`int`	Number of items per page.	required
`total`	`int`	Total number of matching conversations.	required

items `instance-attribute` ¶

items

page `instance-attribute` ¶

page

per_page `instance-attribute` ¶

per_page

total `instance-attribute` ¶

total

JoinedConversationRow `dataclass` ¶

JoinedConversationRow(conversation, extras)

Row from a paginated conversation listing that joins other tables.

Holds the conversation entity plus values from joined tables (extras), keyed by the column name the UI uses.

conversation `instance-attribute` ¶

conversation

extras `instance-attribute` ¶

extras

review_form ¶

ConditionalRequirement `dataclass` ¶

ConditionalRequirement(field_id, values)

Bases: DataClassJsonMixin

Defines when a field is required based on another field's value.

field_id `instance-attribute` ¶

field_id

values `instance-attribute` ¶

values

ConditionalVisibility `dataclass` ¶

ConditionalVisibility(field_id, contains_value)

Bases: DataClassJsonMixin

Defines when a field should be visible based on another field's value.

contains_value `instance-attribute` ¶

contains_value

field_id `instance-attribute` ¶

field_id

MultiSelectField `dataclass` ¶

MultiSelectField(
    id,
    label,
    field_type,
    required=False,
    options=list(),
    placeholder=None,
    description=None,
    conditional_on=None,
    required_when=None,
    allow_create=False,
)

Bases: ReviewField

A multi-select checkbox/tag field.

__post_init__ ¶

__post_init__()

Source code in components/ai_tooling/public/entities/review_form.py

def __post_init__(self) -> None:  # noqa: D105
    self.field_type = ReviewFieldType.MULTI_SELECT

options `class-attribute` `instance-attribute` ¶

options = field(default_factory=list)

RatingField `dataclass` ¶

RatingField(
    id,
    label,
    field_type,
    required=False,
    options=None,
    placeholder=None,
    description=None,
    conditional_on=None,
    required_when=None,
    allow_create=False,
    min_value=1,
    max_value=5,
)

Bases: ReviewField

A numeric rating field with min/max bounds.

__post_init__ ¶

__post_init__()

Source code in components/ai_tooling/public/entities/review_form.py

def __post_init__(self) -> None:  # noqa: D105
    self.field_type = ReviewFieldType.RATING

max_value `class-attribute` `instance-attribute` ¶

max_value = 5

min_value `class-attribute` `instance-attribute` ¶

min_value = 1

ReviewField `dataclass` ¶

ReviewField(
    id,
    label,
    field_type,
    required=False,
    options=None,
    placeholder=None,
    description=None,
    conditional_on=None,
    required_when=None,
    allow_create=False,
)

Bases: DataClassJsonMixin

A single field in a review form.

This base class includes all possible field attributes for backward compatibility. Use the specialized subclasses (RatingField, TextField, SelectField, MultiSelectField) when you need type-specific defaults and validation.

allow_create `class-attribute` `instance-attribute` ¶

allow_create = False

conditional_on `class-attribute` `instance-attribute` ¶

conditional_on = None

description `class-attribute` `instance-attribute` ¶

description = None

field_type `instance-attribute` ¶

field_type

id `instance-attribute` ¶

id

label `instance-attribute` ¶

label

options `class-attribute` `instance-attribute` ¶

options = None

placeholder `class-attribute` `instance-attribute` ¶

placeholder = None

required `class-attribute` `instance-attribute` ¶

required = False

required_when `class-attribute` `instance-attribute` ¶

required_when = None

ReviewFieldOption `dataclass` ¶

ReviewFieldOption(value, label, group=None)

Bases: DataClassJsonMixin

An option for select-type review fields.

group `class-attribute` `instance-attribute` ¶

group = None

label `instance-attribute` ¶

label

value `instance-attribute` ¶

value

ReviewFieldType ¶

Bases: AlanBaseEnum

Types of fields that can appear in a review form.

MULTI_SELECT `class-attribute` `instance-attribute` ¶

MULTI_SELECT = 'multi_select'

RATING `class-attribute` `instance-attribute` ¶

RATING = 'rating'

SELECT `class-attribute` `instance-attribute` ¶

SELECT = 'select'

TEXT `class-attribute` `instance-attribute` ¶

TEXT = 'text'

TOOL_CALLS_REVIEW `class-attribute` `instance-attribute` ¶

TOOL_CALLS_REVIEW = 'tool_calls_review'

ReviewForm `dataclass` ¶

ReviewForm(fields, submit_label='Submit Review')

Bases: DataClassJsonMixin

A review form configuration.

fields `instance-attribute` ¶

fields

from_ai_review_form_model `staticmethod` ¶

from_ai_review_form_model(form_model)

Convert an AIReviewForm ORM model to a ReviewForm dataclass.

Source code in components/ai_tooling/public/entities/review_form.py

@staticmethod
def from_ai_review_form_model(form_model: AIReviewForm) -> "ReviewForm":
    """Convert an AIReviewForm ORM model to a ReviewForm dataclass."""
    questions = form_model.questions
    fields_data = questions.get("fields", [])
    submit_label = questions.get("submit_label", "Submit Review")

    fields = [review_field_from_dict(field_data) for field_data in fields_data]

    return ReviewForm(fields=fields, submit_label=submit_label)

submit_label `class-attribute` `instance-attribute` ¶

submit_label = 'Submit Review'

SelectField `dataclass` ¶

SelectField(
    id,
    label,
    field_type,
    required=False,
    options=list(),
    placeholder=None,
    description=None,
    conditional_on=None,
    required_when=None,
    allow_create=False,
)

Bases: ReviewField

A single-select dropdown/radio field.

__post_init__ ¶

__post_init__()

Source code in components/ai_tooling/public/entities/review_form.py

def __post_init__(self) -> None:  # noqa: D105
    self.field_type = ReviewFieldType.SELECT

options `class-attribute` `instance-attribute` ¶

options = field(default_factory=list)

TextField `dataclass` ¶

TextField(
    id,
    label,
    field_type,
    required=False,
    options=None,
    placeholder=None,
    description=None,
    conditional_on=None,
    required_when=None,
    allow_create=False,
)

Bases: ReviewField

A free-text input field.

__post_init__ ¶

__post_init__()

Source code in components/ai_tooling/public/entities/review_form.py

def __post_init__(self) -> None:  # noqa: D105
    self.field_type = ReviewFieldType.TEXT

ToolCallsReviewData `dataclass` ¶

ToolCallsReviewData(
    actual_tool_calls=list(), available_tools=list()
)

Bases: DataClassJsonMixin

Data for tool calls review field.

actual_tool_calls `class-attribute` `instance-attribute` ¶

actual_tool_calls = field(default_factory=list)

available_tools `class-attribute` `instance-attribute` ¶

available_tools = field(default_factory=list)

ToolCallsReviewField `dataclass` ¶

ToolCallsReviewField(
    id,
    label,
    field_type,
    required=False,
    options=None,
    placeholder=None,
    description=None,
    conditional_on=None,
    required_when=None,
    allow_create=False,
    tool_calls_data=None,
)

Bases: ReviewField

A field for reviewing tool calls made during a conversation.

__post_init__ ¶

__post_init__()

Source code in components/ai_tooling/public/entities/review_form.py

def __post_init__(self) -> None:  # noqa: D105
    self.field_type = ReviewFieldType.TOOL_CALLS_REVIEW

tool_calls_data `class-attribute` `instance-attribute` ¶

tool_calls_data = None

review_field_from_dict ¶

review_field_from_dict(field_data)

Convert a field dictionary to the appropriate ReviewField subclass.

Source code in components/ai_tooling/public/entities/review_form.py

def review_field_from_dict(field_data: dict[str, Any]) -> ReviewField:
    """Convert a field dictionary to the appropriate ReviewField subclass."""
    field_type = ReviewFieldType(field_data["field_type"])
    conditional_on = _parse_conditional(field_data)
    required_when = _parse_required_when(field_data)

    field_id = field_data["id"]
    label = field_data["label"]
    required = field_data.get("required", False)
    description = field_data.get("description")
    placeholder = field_data.get("placeholder")
    options = _parse_options(field_data) if field_data.get("options") else None

    if field_type == ReviewFieldType.RATING:
        return RatingField(
            id=field_id,
            label=label,
            field_type=field_type,
            required=required,
            description=description,
            conditional_on=conditional_on,
            required_when=required_when,
            min_value=field_data.get("min_value", 1),
            max_value=field_data.get("max_value", 5),
            options=options or [],
        )

    if field_type == ReviewFieldType.TEXT:
        return TextField(
            id=field_id,
            label=label,
            field_type=field_type,
            required=required,
            description=description,
            conditional_on=conditional_on,
            required_when=required_when,
            placeholder=placeholder,
        )

    if field_type == ReviewFieldType.SELECT:
        return SelectField(
            id=field_id,
            label=label,
            field_type=field_type,
            required=required,
            description=description,
            conditional_on=conditional_on,
            required_when=required_when,
            options=options or [],
        )

    if field_type == ReviewFieldType.MULTI_SELECT:
        return MultiSelectField(
            id=field_id,
            label=label,
            field_type=field_type,
            required=required,
            description=description,
            conditional_on=conditional_on,
            required_when=required_when,
            options=options or [],
            allow_create=field_data.get("allow_create", False),
        )

    if field_type == ReviewFieldType.TOOL_CALLS_REVIEW:
        return ToolCallsReviewField(
            id=field_id,
            label=label,
            field_type=field_type,
            required=required,
            description=description,
            conditional_on=conditional_on,
            required_when=required_when,
            tool_calls_data=None,
        )

review_response ¶

Public entity for review response results.

ReviewResponseResult `dataclass` ¶

ReviewResponseResult(
    id, entity_id, reviewer_id, agent_type, answers
)

Bases: DataClassJsonMixin

Result of submitting a review response.

This is the public representation of a submitted review, containing only the data needed by consumers.

Parameters:

Name	Type	Description	Default
`id`	`UUID`	The unique identifier of the created response.	required
`entity_id`	`UUID`	The conversation or response ID that was reviewed.	required
`reviewer_id`	`str`	Profile ID of the Alaner who submitted the review.	required
`agent_type`	`str`	The type of AI agent that was reviewed.	required
`answers`	`dict[str, Any]`	The submitted answers.	required

agent_type `instance-attribute` ¶

agent_type

answers `instance-attribute` ¶

answers

entity_id `instance-attribute` ¶

entity_id

id `instance-attribute` ¶

id

reviewer_id `instance-attribute` ¶

reviewer_id

components.ai_tooling.public.enums ¶

agent_type ¶

Public agent type enum for AI tooling.

AgentType ¶

Bases: AlanBaseEnum

Type of agent being reviewed on AI conversations.

AUTOMATED_RESOLUTION `class-attribute` `instance-attribute` ¶

AUTOMATED_RESOLUTION = 'automated_resolution'

HARRY_CLASSIFIER `class-attribute` `instance-attribute` ¶

HARRY_CLASSIFIER = 'harry_classifier'

HARRY_COMPOSER `class-attribute` `instance-attribute` ¶

HARRY_COMPOSER = 'harry_composer'

config_section ¶

ConfigSection ¶

Bases: AlanBaseEnum

Known config sections used throughout the agent config system.

agents `class-attribute` `instance-attribute` ¶

agents = 'agents'

llm_prompt_templates `class-attribute` `instance-attribute` ¶

llm_prompt_templates = 'llm_prompt_templates'

member_attributes `class-attribute` `instance-attribute` ¶

member_attributes = 'member_attributes'

redirections `class-attribute` `instance-attribute` ¶

redirections = 'redirections'

tools `class-attribute` `instance-attribute` ¶

tools = 'tools'

review_type ¶

Public review type enum for AI tooling.

ReviewType ¶

Bases: AlanBaseEnum

Type of review being performed on AI conversations.

Used to identify different review workflows (monitoring, accuracy checks, etc).

ACCURACY `class-attribute` `instance-attribute` ¶

ACCURACY = 'accuracy'

ANNOTATION `class-attribute` `instance-attribute` ¶

ANNOTATION = 'annotation'

MONITORING `class-attribute` `instance-attribute` ¶

MONITORING = 'monitoring'

components.ai_tooling.public.git_config_store ¶

GitConfigStore ¶

GitConfigStore(
    repo_full_name,
    registry_base,
    prompts_base,
    agent_registry_paths,
    tool_definition_paths=(),
    default_branch="main",
    max_fetch_workers=8,
)

Git-backed agent-config store

The repo is the source of truth: each agent's config is assembled on the fly from its registry YAML + the prompt file it references, read at a given branch/commit. A "version" is a commit SHA.

Source code in components/ai_tooling/public/git_config_store.py

def __init__(
    self,
    repo_full_name: str,
    registry_base: str,
    prompts_base: str,
    agent_registry_paths: dict[str, str],
    tool_definition_paths: Sequence[str] = (),
    default_branch: str = "main",
    max_fetch_workers: int = 8,
) -> None:
    self._repo_full_name = repo_full_name
    self._registry_base = registry_base
    self._prompts_base = prompts_base
    self._agent_registry_paths = agent_registry_paths
    self._tool_definition_paths = tool_definition_paths
    self._default_branch = default_branch
    self._max_fetch_workers = max_fetch_workers

copy ¶

copy(
    source_env,
    source_branch,
    target_env,
    target_branch,
    metadata,
    source_version=None,
)

Create target_branch at the source commit (branch creation only).

Source code in components/ai_tooling/public/git_config_store.py

def copy(
    self,
    source_env: str | None,
    source_branch: str,
    target_env: str | None,  # noqa: ARG002 (git branches are env-agnostic)
    target_branch: str,
    metadata: AgentConfigCommitMetadata,  # noqa: ARG002 (no commit is made)
    source_version: str | None = None,
) -> None:
    """Create `target_branch` at the source commit (branch creation only)."""
    sha = source_version or self.get_latest_version(source_env, source_branch)
    if sha is None:
        raise ValueError(f"Source branch '{source_branch}' has no commits.")
    try:
        self._repo(write=True).create_git_ref(
            ref=f"refs/heads/{target_branch}", sha=sha
        )
    except GithubException as e:
        if e.status == 422:  # GitHub: "Reference already exists"
            raise ValueError(f"Branch '{target_branch}' already exists.") from e
        raise

delete_branch ¶

delete_branch(env, branch)

Unsupported: edits to the git-backed config go through a PR.

Source code in components/ai_tooling/public/git_config_store.py

def delete_branch(self, env: str, branch: str) -> None:
    """Unsupported: edits to the git-backed config go through a PR."""
    raise NotImplementedError("GitConfigStore is read-only")

fetch_agent ¶

fetch_agent(name, env, branch, version)

Assemble a single agent's sections + commit metadata at version.

Cheaper than fetch_all: reads only this agent's registry/prompt/tool files (plus the shared tool-definition files), not every agent's.

Source code in components/ai_tooling/public/git_config_store.py

def fetch_agent(
    self,
    name: str,
    env: str,  # noqa: ARG002
    branch: str,
    version: str,
) -> dict[str, Any]:
    """Assemble a single agent's sections + commit metadata at `version`.

    Cheaper than `fetch_all`: reads only this agent's registry/prompt/tool
    files (plus the shared tool-definition files), not every agent's.
    """
    registry_path = self._agent_registry_paths.get(name)
    if registry_path is None:
        raise ValueError(f"Unknown agent '{name}'")
    repo = self._repo()
    ref = version or branch
    # The agent config, tool definitions, and commit metadata are independent
    # GitHub reads — fetch them concurrently instead of serially.
    with ThreadPoolExecutor(max_workers=3) as pool:
        config_future = pool.submit(
            self._build_agent_config, repo, ref, name, registry_path
        )
        definitions_future = pool.submit(self._load_tool_definitions, repo, ref)
        metadata_future = pool.submit(self._commit_metadata, repo, ref, branch)
        config = config_future.result()
        tool_definitions = definitions_future.result()
        metadata = metadata_future.result()
    return {
        ConfigSection.agents.value: [self._agent_entry(config)],
        ConfigSection.llm_prompt_templates.value: [
            self._agent_prompt_template(config, ref)
        ]
        + self._included_prompt_templates([config], ref),
        ConfigSection.tools.value: self._tools_catalog([config], tool_definitions),
        "metadata": metadata.to_dict(),
    }

fetch_all ¶

fetch_all(env, branch, version)

Assemble all sections + commit metadata at version (a commit SHA).

Source code in components/ai_tooling/public/git_config_store.py

def fetch_all(self, env: str, branch: str, version: str) -> dict[str, Any]:  # noqa: ARG002
    """Assemble all sections + commit metadata at `version` (a commit SHA)."""
    repo = self._repo()
    ref = version or branch
    agent_configs = self._build_agent_configs(repo, ref)
    tool_definitions = self._load_tool_definitions(repo, ref)
    return {
        ConfigSection.agents.value: [
            self._agent_entry(config) for config in agent_configs
        ],
        ConfigSection.llm_prompt_templates.value: [
            self._agent_prompt_template(config, ref) for config in agent_configs
        ]
        + self._included_prompt_templates(agent_configs, ref),
        ConfigSection.tools.value: self._tools_catalog(
            agent_configs, tool_definitions
        ),
        "metadata": self._commit_metadata(repo, ref, branch).to_dict(),
    }

fetch_recent_commits ¶

fetch_recent_commits(env, branch, version, limit=10)

Most recent commits on branch, newest first, as commit metadata.

Source code in components/ai_tooling/public/git_config_store.py

def fetch_recent_commits(
    self,
    env: str,  # noqa: ARG002
    branch: str,
    version: str,  # noqa: ARG002
    limit: int = 10,
) -> list[AgentConfigCommitMetadata]:
    """Most recent commits on `branch`, newest first, as commit metadata."""
    commits = self._repo().get_commits(sha=branch)
    return [
        self._metadata_from_commit(commit, branch) for commit in commits[:limit]
    ]

fetch_review_state ¶

fetch_review_state(env, branch, version)

Always None: the git-backed config has no peer-review workflow.

Source code in components/ai_tooling/public/git_config_store.py

def fetch_review_state(
    self,
    env: str,  # noqa: ARG002
    branch: str,  # noqa: ARG002
    version: str,  # noqa: ARG002
) -> ReviewState | None:
    """Always None: the git-backed config has no peer-review workflow."""
    return None

fetch_section ¶

fetch_section(section, env, branch, version)

Return a single section's assembled data at version.

Source code in components/ai_tooling/public/git_config_store.py

def fetch_section(self, section: str, env: str, branch: str, version: str) -> Any:
    """Return a single section's assembled data at `version`."""
    key = getattr(section, "value", section)
    return self.fetch_all(env, branch, version).get(key)

fetch_tools_catalog ¶

fetch_tools_catalog(env, branch, version)

Full catalog of every defined tool, read straight from the tool-definition files (no agent fan-out).

Source code in components/ai_tooling/public/git_config_store.py

def fetch_tools_catalog(
    self,
    env: str,  # noqa: ARG002
    branch: str,
    version: str,
) -> list[dict[str, Any]]:
    """Full catalog of every defined tool, read straight from the tool-definition
    files (no agent fan-out).
    """
    repo = self._repo()
    ref = version or branch
    definitions = self._load_tool_definitions(repo, ref)
    return [
        {
            "name": name,
            "description": definition.get("description", ""),
            "params": definition.get("params", []),
        }
        for name, definition in sorted(definitions.items())
    ]

for_country ¶

for_country(app_name=None)

No-op: the git-backed config is global, not partitioned by country.

Source code in components/ai_tooling/public/git_config_store.py

def for_country(self, app_name: AppName | None = None) -> "GitConfigStore":  # noqa: ARG002
    """No-op: the git-backed config is global, not partitioned by country."""
    return self

get_latest_version ¶

get_latest_version(env, branch)

HEAD commit SHA of branch, or None if the branch doesn't exist.

Source code in components/ai_tooling/public/git_config_store.py

def get_latest_version(self, env: str | None, branch: str) -> str | None:  # noqa: ARG002
    """HEAD commit SHA of `branch`, or None if the branch doesn't exist."""
    try:
        return self._repo().get_branch(branch).commit.sha
    except (GithubException, UnknownObjectException):
        return None

list_agent_names ¶

list_agent_names(env, branch, version)

The agent names, straight from the registry config (no GitHub read).

Source code in components/ai_tooling/public/git_config_store.py

def list_agent_names(self, env: str, branch: str, version: str) -> list[str]:  # noqa: ARG002
    """The agent names, straight from the registry config (no GitHub read)."""
    return list(self._agent_registry_paths.keys())

list_branches ¶

list_branches(env)

The git-backed config only exposes its configured default branch.

Source code in components/ai_tooling/public/git_config_store.py

def list_branches(self, env: str) -> list[str]:  # noqa: ARG002
    """The git-backed config only exposes its configured default branch."""
    return [self._default_branch]

remove_version ¶

remove_version(env, branch, version)

Unsupported: edits to the git-backed config go through a PR.

Source code in components/ai_tooling/public/git_config_store.py

def remove_version(self, env: str, branch: str, version: str) -> None:
    """Unsupported: edits to the git-backed config go through a PR."""
    raise NotImplementedError("GitConfigStore is read-only")

rollback ¶

rollback(target_version, env, branch, metadata)

Unsupported: edits to the git-backed config go through a PR.

Source code in components/ai_tooling/public/git_config_store.py

def rollback(
    self,
    target_version: str,
    env: str,
    branch: str,
    metadata: AgentConfigCommitMetadata,
) -> None:
    """Unsupported: edits to the git-backed config go through a PR."""
    raise NotImplementedError("GitConfigStore is read-only")

sections `class-attribute` `instance-attribute` ¶

sections = (agents, llm_prompt_templates, tools)

upload_all_sections ¶

upload_all_sections(
    sections_data, metadata, env, branch, version
)

Unsupported: edits to the git-backed config go through a PR.

Source code in components/ai_tooling/public/git_config_store.py

def upload_all_sections(
    self,
    sections_data: dict[ConfigSection, Any],
    metadata: AgentConfigCommitMetadata,
    env: str,
    branch: str,
    version: str,
) -> None:
    """Unsupported: edits to the git-backed config go through a PR."""
    raise NotImplementedError("GitConfigStore is read-only")

upload_review_state ¶

upload_review_state(data, env, branch, version)

Unsupported: edits to the git-backed config go through a PR.

Source code in components/ai_tooling/public/git_config_store.py

def upload_review_state(
    self, data: ReviewState, env: str, branch: str, version: str
) -> None:
    """Unsupported: edits to the git-backed config go through a PR."""
    raise NotImplementedError("GitConfigStore is read-only")

components.ai_tooling.public.github_app ¶

agent_studio_github_app_client ¶

agent_studio_github_app_client(write=False)

Build a GithubClient authed as the Agent Studio GitHub App.

write=True mints a token that can also create branches/commits/PRs; otherwise the token is read-only. In test mode, returns a fake-token client (callers mock the GitHub API).

Source code in components/ai_tooling/public/github_app.py

def agent_studio_github_app_client(write: bool = False) -> GithubClient:
    """Build a `GithubClient` authed as the Agent Studio GitHub App.

    `write=True` mints a token that can also create branches/commits/PRs;
    otherwise the token is read-only. In test mode, returns a fake-token client
    (callers mock the GitHub API).
    """
    if is_test_mode():
        return GithubClient(token=_FAKE_TEST_TOKEN)
    app_id = current_config.get(_APP_ID)
    installation_id = current_config.get(_INSTALLATION_ID)
    if not app_id or not installation_id:
        raise ValueError(
            f"{_APP_ID} / {_INSTALLATION_ID} are not configured for this app."
        )
    private_key = raw_secret_from_config(_PRIVATE_KEY_SECRET_NAME)
    if not private_key:
        raise ValueError(f"{_PRIVATE_KEY_SECRET_NAME} secret is missing or empty.")
    return GithubClient.for_github_app(
        app_id=app_id,
        private_key=private_key,
        installation_id=installation_id,
        permissions=_WRITE_PERMISSIONS if write else _READ_ONLY_PERMISSIONS,
    )

components.ai_tooling.public.helpers ¶

list_item_builders ¶

create_flask_admin_list_item ¶

create_flask_admin_list_item(
    model_name, entity_id, base_url
)

Create a link to the item in Flask Admin

Source code in components/ai_tooling/public/helpers/list_item_builders.py

def create_flask_admin_list_item(
    model_name: str, entity_id: str, base_url: str
) -> ListItem:
    """
    Create a link to the item in Flask Admin
    """
    return ListItem(
        text=f"Flask Admin: {model_name}",
        link_url=f"{base_url}/admin/{model_name.lower()}/details/?id={entity_id}",
        copy_value=entity_id,
        copy_label=f"Copy {model_name} ID",
    )

create_intercom_list_item ¶

create_intercom_list_item(
    intercom_conversation_id, intercom_workspace_id
)

Create a link to the Intercom conversation

Source code in components/ai_tooling/public/helpers/list_item_builders.py

def create_intercom_list_item(
    intercom_conversation_id: str | None, intercom_workspace_id: str
) -> ListItem:
    """
    Create a link to the Intercom conversation
    """
    if intercom_conversation_id:
        return ListItem(
            text=f"Intercom ({intercom_conversation_id})",
            link_url=f"https://app.eu.intercom.com/a/inbox/{intercom_workspace_id}/inbox/shared/all/conversation/{intercom_conversation_id}",
            copy_value=intercom_conversation_id,
            copy_label="Copy Intercom conversation ID",
        )
    else:
        return ListItem(text="No Intercom conversation id")

create_marmot_user_list_item ¶

create_marmot_user_list_item(
    user_id, frontend_base_url, country="fr"
)

Create a link to the user's Marmot profile

Source code in components/ai_tooling/public/helpers/list_item_builders.py

def create_marmot_user_list_item(
    user_id: str | None,
    frontend_base_url: str,
    country: str = "fr",
) -> ListItem:
    """
    Create a link to the user's Marmot profile
    """
    if user_id:
        return ListItem(
            text=f"Marmot profile ({user_id})",
            link_url=f"{frontend_base_url}/marmot/{country}/user/{user_id}",
            copy_value=user_id,
            copy_label="Copy Marmot User ID",
        )
    else:
        return ListItem(text="No Marmot profile")

components.ai_tooling.public.queries ¶

review_form ¶

Public queries for review form retrieval.

get_review_form_by_type ¶

get_review_form_by_type(agent_type, review_type)

Fetch the latest review form configuration for a given agent and review type.

Parameters:

Name	Type	Description	Default
`agent_type`	`str`	The type of AI agent (e.g., "automated_resolution").	required
`review_type`	`ReviewType`	The type of review workflow.	required

Returns:

Type	Description
`ReviewForm \| None`	A ReviewForm if found, None otherwise.

Source code in components/ai_tooling/public/queries/review_form.py

def get_review_form_by_type(
    agent_type: str, review_type: ReviewType
) -> ReviewForm | None:
    """Fetch the latest review form configuration for a given agent and review type.

    Args:
        agent_type: The type of AI agent (e.g., "automated_resolution").
        review_type: The type of review workflow.

    Returns:
        A ReviewForm if found, None otherwise.
    """
    stmt = (
        select(AIReviewForm)
        .where(
            AIReviewForm.agent_type == agent_type,
            AIReviewForm.review_type == review_type.value,
        )
        .order_by(AIReviewForm.version.desc())
        .limit(1)
    )

    form_model = current_session.execute(stmt).scalar_one_or_none()

    if form_model is None:
        return None

    return ReviewForm.from_ai_review_form_model(form_model)

components.ai_tooling.public.s3_config_store ¶

AGENT_CONFIGS_S3_BUCKET `module-attribute` ¶

AGENT_CONFIGS_S3_BUCKET = 'ai-agent-configs'

S3ConfigStore ¶

S3ConfigStore(prefix, sections)

S3 config store.

S3 key layout

{prefix}/{env}/{branch}/{version}/{section}.json {prefix}/{env}/{branch}/commit_history.txt

Parameters:

Name	Type	Description	Default
`prefix`	`str`	Path prefix isolating this deployment (e.g. "automated-resolution", "harry").	required
`sections`	`Sequence[ConfigSection]`	Section names this store manages (e.g. ["agents", "llm_prompt_templates"]).	required

Source code in components/ai_tooling/public/s3_config_store.py

def __init__(
    self,
    prefix: str,
    sections: Sequence[ConfigSection],
) -> None:
    self.prefix = prefix
    self.sections = sections

append_to_commit_history ¶

append_to_commit_history(version, env, branch)

Append a version to the commit history file.

Source code in components/ai_tooling/public/s3_config_store.py

def append_to_commit_history(self, version: str, env: str, branch: str) -> None:
    """Append a version to the commit history file."""
    existing = self.fetch_commit_history(env, branch)
    existing.append(version)
    self._upload_commit_history(existing, env, branch)

copy ¶

copy(
    source_env,
    source_branch,
    target_env,
    target_branch,
    metadata,
    source_version=None,
)

Copy source configs to target as a new commit.

Source code in components/ai_tooling/public/s3_config_store.py

def copy(
    self,
    source_env: str | None,
    source_branch: str,
    target_env: str | None,
    target_branch: str,
    metadata: AgentConfigCommitMetadata,
    source_version: str | None = None,
) -> None:
    """Copy source configs to target as a new commit."""
    # S3 storage is partitioned by env, so it is mandatory here.
    if source_env is None or target_env is None:
        raise ValueError("source_env and target_env are required for S3 storage.")

    source_version = source_version or self.get_latest_version(
        source_env, source_branch
    )
    if source_version is None:
        raise ValueError(
            f"Cannot copy: source branch {source_branch} has no versions"
        )

    source_data = self.fetch_all(source_env, source_branch, source_version)

    sections_data: dict[ConfigSection, Any] = {
        s: source_data[s] for s in self.sections
    }

    self.upload_all_sections(
        sections_data, metadata, target_env, target_branch, metadata.created_at
    )

delete_branch ¶

delete_branch(env, branch)

Delete all S3 objects for a branch.

Source code in components/ai_tooling/public/s3_config_store.py

def delete_branch(self, env: str, branch: str) -> None:
    """Delete all S3 objects for a branch."""
    if branch == "main":
        raise ValueError("Cannot delete the main branch")
    prefix = f"{self.prefix}/{env}/{branch}/"
    get_boto3_session().resource("s3").Bucket(
        AGENT_CONFIGS_S3_BUCKET
    ).objects.filter(Prefix=prefix).delete()

fetch_agent ¶

fetch_agent(name, env, branch, version)

Fetch all sections with the agents section narrowed to a single agent.

Source code in components/ai_tooling/public/s3_config_store.py

def fetch_agent(
    self, name: str, env: str, branch: str, version: str
) -> dict[str, Any]:
    """Fetch all sections with the `agents` section narrowed to a single agent."""
    configs = self.fetch_all(env, branch, version)
    agents = configs.get(ConfigSection.agents.value, [])
    agent = next((a for a in agents if a.get("name") == name), None)
    if agent is None:
        raise Exception(f"Unknown agent '{name}'")
    return {**configs, ConfigSection.agents.value: [agent]}

fetch_all ¶

fetch_all(env, branch, version)

Fetch all sections + commit metadata from S3 in parallel.

Returns {section_name: section_json, "metadata": commit_metadata_dict}.

Source code in components/ai_tooling/public/s3_config_store.py

def fetch_all(self, env: str, branch: str, version: str) -> dict[str, Any]:
    """Fetch all sections + commit metadata from S3 in parallel.

    Returns `{section_name: section_json, "metadata": commit_metadata_dict}`.
    """
    with ThreadPoolExecutor(max_workers=len(self.sections) + 1) as executor:
        section_futures = {
            executor.submit(self.fetch_section, s, env, branch, version): s
            for s in self.sections
        }
        metadata_future = executor.submit(
            self._fetch_commit_metadata, env, branch, version
        )

        results: dict[str, Any] = {}
        for future in as_completed(section_futures):
            results[section_futures[future]] = future.result()
        results[_METADATA_SECTION] = metadata_future.result().to_dict()

    return results

fetch_commit_history ¶

fetch_commit_history(env, branch)

Fetch the commit history (list of version strings) from S3.

Returns empty list if the commit history file doesn't exist.

Source code in components/ai_tooling/public/s3_config_store.py

def fetch_commit_history(self, env: str, branch: str) -> list[str]:
    """Fetch the commit history (list of version strings) from S3.

    Returns empty list if the commit history file doesn't exist.
    """
    s3_key = self._commit_history_key(env, branch)
    try:
        s3_obj = RemoteFileClient.client().get_object(
            Bucket=AGENT_CONFIGS_S3_BUCKET, Key=s3_key
        )
        content = s3_obj["Body"].read().decode("utf-8")
        return [line.strip() for line in content.split("\n") if line.strip()]
    except ClientError as e:
        if e.response["Error"]["Code"] == "NoSuchKey":
            return []
        raise

fetch_recent_commits ¶

fetch_recent_commits(env, branch, version, limit=10)

Fetch the most recent commits up to the given version, newest first.

Source code in components/ai_tooling/public/s3_config_store.py

def fetch_recent_commits(
    self, env: str, branch: str, version: str, limit: int = 10
) -> list[AgentConfigCommitMetadata]:
    """Fetch the most recent commits up to the given version, newest first."""
    full_history = self.fetch_commit_history(env, branch)
    if version not in full_history:
        current_logger.warning(
            "Version not found in commit history", version=version, branch=branch
        )
        return []
    end = full_history.index(version) + 1
    recent = list(reversed(full_history[max(0, end - limit) : end]))

    with ThreadPoolExecutor(max_workers=min(len(recent), 5)) as executor:
        futures = [
            executor.submit(self._fetch_commit_metadata, env, branch, v)
            for v in recent
        ]

    commits: list[AgentConfigCommitMetadata] = []
    for v, future in zip(recent, futures):
        try:
            commits.append(future.result())
        except Exception:
            current_logger.warning("Failed to load commit metadata", version=v)
    return commits

fetch_review_state ¶

fetch_review_state(env, branch, version)

Fetch review_state.json for a version, or None if it doesn't exist.

Source code in components/ai_tooling/public/s3_config_store.py

def fetch_review_state(
    self, env: str, branch: str, version: str
) -> ReviewState | None:
    """Fetch review_state.json for a version, or None if it doesn't exist."""
    s3_key = self._section_key("review_state", env, branch, version)
    try:
        s3_obj = RemoteFileClient.client().get_object(
            Bucket=AGENT_CONFIGS_S3_BUCKET, Key=s3_key
        )
        return ReviewState.from_json(s3_obj["Body"].read().decode("utf-8"))
    except ClientError as e:
        if e.response["Error"]["Code"] == "NoSuchKey":
            return None
        raise

fetch_section ¶

fetch_section(section, env, branch, version)

Fetch a single section's JSON from S3.

Source code in components/ai_tooling/public/s3_config_store.py

def fetch_section(self, section: str, env: str, branch: str, version: str) -> Any:
    """Fetch a single section's JSON from S3."""
    s3_key = self._section_key(section, env, branch, version)
    try:
        s3_obj = RemoteFileClient.client().get_object(
            Bucket=AGENT_CONFIGS_S3_BUCKET, Key=s3_key
        )
        return json.loads(s3_obj["Body"].read().decode("utf-8"))
    except ClientError as e:
        if e.response["Error"]["Code"] == "NoSuchKey":
            raise Exception(
                f"Config {section}.json not found for version {version} branch {branch}"
            ) from e
        raise

fetch_tools_catalog ¶

fetch_tools_catalog(env, branch, version)

Return the persisted tools section (empty when tools are injected by an enricher rather than stored).

Source code in components/ai_tooling/public/s3_config_store.py

def fetch_tools_catalog(
    self, env: str, branch: str, version: str
) -> list[dict[str, Any]]:
    """Return the persisted `tools` section (empty when tools are injected by an
    enricher rather than stored).
    """
    configs = self.fetch_all(env, branch, version)
    return configs.get(ConfigSection.tools.value, [])

for_country ¶

for_country(app_name=None)

Return a new store with the country-specific prefix for country-specific storage.

Source code in components/ai_tooling/public/s3_config_store.py

def for_country(self, app_name: AppName | None = None) -> "S3ConfigStore":
    """Return a new store with the country-specific prefix for country-specific storage."""
    country_prefix = Country.from_app_name(app_name or get_current_app_name())

    return S3ConfigStore(
        prefix=f"{self.prefix}/{country_prefix}",
        sections=self.sections,
    )

get_latest_version ¶

get_latest_version(env, branch)

Get the latest version from commit history, or None if the branch has no versions.

Source code in components/ai_tooling/public/s3_config_store.py

def get_latest_version(self, env: str, branch: str) -> str | None:
    """Get the latest version from commit history, or None if the branch has no versions."""
    lines = self.fetch_commit_history(env, branch)
    return lines[-1] if lines else None

list_agent_names ¶

list_agent_names(env, branch, version)

List agent names from the agents section.

Source code in components/ai_tooling/public/s3_config_store.py

def list_agent_names(self, env: str, branch: str, version: str) -> list[str]:
    """List agent names from the `agents` section."""
    agents = (
        self.fetch_section(ConfigSection.agents.value, env, branch, version) or []
    )
    return [a["name"] for a in agents if a.get("name")]

list_branches ¶

list_branches(env)

List all branches for an environment.

Source code in components/ai_tooling/public/s3_config_store.py

def list_branches(self, env: str) -> list[str]:
    """List all branches for an environment."""
    prefix = self._branches_prefix(env)
    response = RemoteFileClient.client().list_objects_v2(
        Bucket=AGENT_CONFIGS_S3_BUCKET, Prefix=prefix, Delimiter="/"
    )
    common_prefixes = response.get("CommonPrefixes", [])
    # Prefix looks like "{self.prefix}/{env}/{branch}/" — extract branch
    prefix_depth = len(prefix.split("/")) - 1  # number of segments before branch
    return [
        item["Prefix"].rstrip("/").split("/")[prefix_depth]
        for item in common_prefixes
    ]

prefix `instance-attribute` ¶

prefix = prefix

remove_version ¶

remove_version(env, branch, version)

Remove a version's files from S3 and its entry from commit history.

Source code in components/ai_tooling/public/s3_config_store.py

def remove_version(self, env: str, branch: str, version: str) -> None:
    """Remove a version's files from S3 and its entry from commit history."""
    s3_key = self._version_prefix(env, branch, version)
    RemoteFileClient.delete(s3_key, s3_bucket=AGENT_CONFIGS_S3_BUCKET)
    self._remove_version_from_commit_history(env, branch, version)

rollback ¶

rollback(target_version, env, branch, metadata)

Rollback by re-publishing an older commit as a new one.

Source code in components/ai_tooling/public/s3_config_store.py

def rollback(
    self,
    target_version: str,
    env: str,
    branch: str,
    metadata: AgentConfigCommitMetadata,
) -> None:
    """Rollback by re-publishing an older commit as a new one."""
    self.copy(
        source_env=env,
        source_branch=branch,
        target_env=env,
        target_branch=branch,
        metadata=metadata,
        source_version=target_version,
    )

sections `instance-attribute` ¶

sections = sections

upload_all_sections ¶

upload_all_sections(
    sections_data, metadata, env, branch, version
)

Publish a commit: upload all sections + commit metadata, then append to commit history.

Source code in components/ai_tooling/public/s3_config_store.py

def upload_all_sections(
    self,
    sections_data: dict[ConfigSection, Any],
    metadata: AgentConfigCommitMetadata,
    env: str,
    branch: str,
    version: str,
) -> None:
    """Publish a commit: upload all sections + commit metadata, then append to commit history."""
    all_uploads: dict[str, Any] = {
        **{s: sections_data[s] for s in self.sections},
        # Strip path-derived fields; they live in the S3 key, not the file.
        _METADATA_SECTION: {
            k: v
            for k, v in metadata.to_dict().items()
            if k not in ("env", "branch", "version")
        },
    }

    with ThreadPoolExecutor(max_workers=len(all_uploads)) as executor:
        futures = [
            executor.submit(
                self.upload_section, data, section, env, branch, version
            )
            for section, data in all_uploads.items()
        ]
        for future in as_completed(futures):
            future.result()

    self.append_to_commit_history(version, env, branch)

upload_review_state ¶

upload_review_state(data, env, branch, version)

Upload review_state.json for a version.

Source code in components/ai_tooling/public/s3_config_store.py

def upload_review_state(
    self, data: ReviewState, env: str, branch: str, version: str
) -> None:
    """Upload review_state.json for a version."""
    s3_key = self._section_key("review_state", env, branch, version)
    json_bytes = data.to_json().encode()
    RemoteFileClient.client().put_object(
        Bucket=AGENT_CONFIGS_S3_BUCKET,
        Key=s3_key,
        Body=json_bytes,
        ContentType="application/json",
    )

upload_section ¶

upload_section(data, section, env, branch, version)

Upload a single section's JSON to S3.

Source code in components/ai_tooling/public/s3_config_store.py

def upload_section(
    self, data: Any, section: str, env: str, branch: str, version: str
) -> None:
    """Upload a single section's JSON to S3."""
    json_bytes = json.dumps(
        data, ensure_ascii=False, default=_json_serializer
    ).encode()
    RemoteFileClient.upload(
        upload_dir=self._version_prefix(env, branch, version),
        file_path_or_file_obj=io.BytesIO(json_bytes),
        s3_bucket=AGENT_CONFIGS_S3_BUCKET,
        destination_filename=f"{section}.json",
    )

with_sub_prefix ¶

with_sub_prefix(suffix)

Return a new store with the suffix appended to the prefix for country-specific storage.

Source code in components/ai_tooling/public/s3_config_store.py

def with_sub_prefix(self, suffix: str) -> "S3ConfigStore":
    """Return a new store with the suffix appended to the prefix for country-specific storage."""
    return S3ConfigStore(
        prefix=f"{self.prefix}/{suffix}",
        sections=self.sections,
    )

Api reference

components.ai_tooling.public.actions ¶

review_response ¶

submit_review_response ¶

components.ai_tooling.public.agent_config_resolver ¶

get_current_user_branch ¶

get_latest_version_from_s3 ¶

get_nonnull_env_branch_version ¶

components.ai_tooling.public.agent_config_stores ¶

ar_config_store module-attribute ¶

harry_config_store module-attribute ¶

legal_complaint_config_store module-attribute ¶

components.ai_tooling.public.ai_agent_config_tool ¶

AIAgentConfigTool ¶

get_llm_prompt_template ¶

get_member_attribute_names ¶

register_smorest_routes ¶

ConfigEnricher module-attribute ¶

standard_sections_enricher ¶

components.ai_tooling.public.ai_debug_tool ¶

AIDebugTool ¶

register_smorest_routes ¶

DebugConversationsGetQuerySchema ¶

filters class-attribute instance-attribute ¶

page class-attribute instance-attribute ¶

per_page class-attribute instance-attribute ¶

components.ai_tooling.public.ai_review_tool ¶

AIReviewTool ¶

register_smorest_routes ¶

AssignConversationResult dataclass ¶

assigned_to_user_id class-attribute instance-attribute ¶

status instance-attribute ¶

AssignmentStatus ¶

ALREADY_ASSIGNED_TO_YOU class-attribute instance-attribute ¶

ASSIGNED class-attribute instance-attribute ¶

CONFLICT class-attribute instance-attribute ¶

ReviewConversationsGetQuerySchema ¶

filters class-attribute instance-attribute ¶

page class-attribute instance-attribute ¶

per_page class-attribute instance-attribute ¶

SubmitAndNextPostJsonSchema ¶

answers class-attribute instance-attribute ¶

filters class-attribute instance-attribute ¶

SubmitAndNextResult dataclass ¶

next_entity_id class-attribute instance-attribute ¶

review_id class-attribute instance-attribute ¶

SubmitReviewCallable module-attribute ¶

SubmitReviewPostJsonSchema ¶

answers class-attribute instance-attribute ¶

SubmitReviewResult dataclass ¶

entity_id instance-attribute ¶

id instance-attribute ¶

components.ai_tooling.public.ai_tooling_app_group ¶

ai_tooling module-attribute ¶

components.ai_tooling.public.ai_tooling_service ¶

AiToolingService ¶

get_llm_prompt_template staticmethod ¶

is_prompt_template_supported staticmethod ¶

resolve_env_branch_version_for_prompt staticmethod ¶

components.ai_tooling.public.blueprint ¶

ai_tooling_blueprint module-attribute ¶

components.ai_tooling.public.commands ¶

seed_agent_config_store ¶

SEEDED_ENVS module-attribute ¶

seed_agent_config_store ¶

components.ai_tooling.public.config_store ¶

ConfigStore ¶

copy ¶

delete_branch ¶

fetch_agent ¶

fetch_all ¶

fetch_recent_commits ¶

fetch_review_state ¶

fetch_section ¶

fetch_tools_catalog ¶

for_country ¶

get_latest_version ¶

list_agent_names ¶

list_branches ¶

remove_version ¶

ar_config_store `module-attribute` ¶

harry_config_store `module-attribute` ¶

legal_complaint_config_store `module-attribute` ¶

ConfigEnricher `module-attribute` ¶

filters `class-attribute` `instance-attribute` ¶

page `class-attribute` `instance-attribute` ¶

per_page `class-attribute` `instance-attribute` ¶

AssignConversationResult `dataclass` ¶

assigned_to_user_id `class-attribute` `instance-attribute` ¶

status `instance-attribute` ¶

ALREADY_ASSIGNED_TO_YOU `class-attribute` `instance-attribute` ¶

ASSIGNED `class-attribute` `instance-attribute` ¶

CONFLICT `class-attribute` `instance-attribute` ¶

filters `class-attribute` `instance-attribute` ¶

page `class-attribute` `instance-attribute` ¶

per_page `class-attribute` `instance-attribute` ¶

answers `class-attribute` `instance-attribute` ¶

filters `class-attribute` `instance-attribute` ¶

SubmitAndNextResult `dataclass` ¶

next_entity_id `class-attribute` `instance-attribute` ¶

review_id `class-attribute` `instance-attribute` ¶

SubmitReviewCallable `module-attribute` ¶

answers `class-attribute` `instance-attribute` ¶

SubmitReviewResult `dataclass` ¶

entity_id `instance-attribute` ¶

id `instance-attribute` ¶

ai_tooling `module-attribute` ¶

get_llm_prompt_template `staticmethod` ¶

is_prompt_template_supported `staticmethod` ¶

resolve_env_branch_version_for_prompt `staticmethod` ¶

ai_tooling_blueprint `module-attribute` ¶

SEEDED_ENVS `module-attribute` ¶

sections `instance-attribute` ¶

REQUIRE_PEER_REVIEW_FEATURE_FLAG `module-attribute` ¶

get_default_redirection_fields `abstractmethod` ¶

get_redirection_screen_names `abstractmethod` ¶

COMPONENT_NAME `module-attribute` ¶

AgentConfigCommitMetadata `dataclass` ¶

author `instance-attribute` ¶

branch `class-attribute` `instance-attribute` ¶

create `classmethod` ¶

created_at `instance-attribute` ¶

env `class-attribute` `instance-attribute` ¶

message `instance-attribute` ¶

review_skipped `class-attribute` `instance-attribute` ¶

version `class-attribute` `instance-attribute` ¶

LlmParams `dataclass` ¶

provider_and_model `instance-attribute` ¶

reasoning_effort `class-attribute` `instance-attribute` ¶

temperature `instance-attribute` ¶

PromptTemplate `dataclass` ¶

id `instance-attribute` ¶

name `instance-attribute` ¶

params `instance-attribute` ¶

prompt_template `instance-attribute` ¶

version `instance-attribute` ¶

ReviewState `dataclass` ¶

commit_message `instance-attribute` ¶

requester `instance-attribute` ¶

reviewer `class-attribute` `instance-attribute` ¶

status `instance-attribute` ¶

version `instance-attribute` ¶

approved `class-attribute` `instance-attribute` ¶

pending `class-attribute` `instance-attribute` ¶

AgentPromptMapping `dataclass` ¶

agent_name `instance-attribute` ¶

prompt_template_names `class-attribute` `instance-attribute` ¶

BaseContextSection `dataclass` ¶

section_type `instance-attribute` ¶

title `instance-attribute` ¶

INFORMATION_LIST `class-attribute` `instance-attribute` ¶

ListContextSection `dataclass` ¶

list_items `instance-attribute` ¶

ListItem `dataclass` ¶

copy_label `class-attribute` `instance-attribute` ¶

copy_value `class-attribute` `instance-attribute` ¶

link_url `class-attribute` `instance-attribute` ¶

text `instance-attribute` ¶

AgentHandoverConversationPart `dataclass` ¶

from_agent `instance-attribute` ¶