keble-helpers 1.40.1

No project description provided

Keble helpers

Just a collection of helper functions used by keble project.

Version 1.40.0 Provider identity projection

KebleProviderId.for_provider(...) projects existing chat-tool or subagent provider id values into the first-class provider enum without changing wire strings. Downstream packages should use this before reading .origin instead of duplicating provider-id string maps.

Version 1.39.2 Client-tool payload alias fix

build_client_action_request(...) now serializes nested payload fields with Pydantic aliases while still removing both tool_type and toolType from the nested payload. The outer request envelope remains {tool_type, payload}.

Version 1.39.1 Additive agent-core contracts

Adds the Stage B Phase 1 agent-core contract surface without changing existing wire shapes:

• client_tool_payload.py owns typed client-tool request/result envelopes and readers. Domain packages can build {tool_type, payload} from typed payloads; backend/core readers can validate envelope structure before parsing domain-specific payloads.
• agent_descriptor.py adds first-class AgentDescriptor, AgentLifecycleResult, AgentContext, AgentToolScope, and provider-manifest schemas while the legacy subagent.py contracts remain importable.
• ArtifactRef and the optional MemoryLink.ref projection let memory records carry user-renderable artifact links beside the existing refId.
• KebleProviderId.origin, KebleOrigin, AgentId, and KebleTaskType.NICHE make provider/task/agent identity alignment explicit without renaming existing provider values.
• AgenticActionProgress now lives in events.py as the event-shaped progress contract; the old client_action.py import path is preserved.

Compatibility rule: this release is additive. Downstream repos may adopt the new canonical modules gradually; cleanup of old aliases belongs to the final cleanup phase after backend, keble-core, and frontend readers are migrated.

Version 1.34.0 Unified Agentic Identity Vocabulary

keble_helpers.ai.identities is the ONE canonical source of truth for the cross-repo agentic identity strings that previously drifted as hardcoded constants in each package:

• KebleTaskType — every task-type routing string. keble.backend's TaskType aliases this enum and Settings.*_TASK_TYPE defaults derive from it.
• SubAgentArchetypeName — every registered archetype id (e.g. positioning_study, segmenting_study, bootstrap_amz_report, product_discovery, product_niche). Backend binding dispatch, package descriptors, and the keble-core frontend mirror all reference these members.
• SubAgentProviderId — every sub-agent / child-query provider id (the compose_subagent_providers dedup key).

All three are str, Enum, so a member is accepted anywhere a plain str is expected (the keble-agentic-chat framework fields stay str); values are typed at the producing edges only. Consuming packages import these instead of re-declaring strings, which makes drift structurally impossible and provider-id collisions catchable by one contract test. Trade-off: adding a new archetype requires adding a member here and a keble-helpers release.

Version 1.28.0 Batched Embedding Helper

aembed_in_batches (keble_helpers.ai.embedding_batching) is the ONE canonical place that chunks a list of texts to a provider's hard per-request cap. Embedding providers reject oversized requests with a permanent HTTP 400 — Azure Cohere embed-v-4-0 caps at 96 texts (total number of texts must be at most 96), which is a request-shape error, not a transient fault, so it must be avoided by chunking, never retried.

from keble_helpers import aembed_in_batches

async def _embed_chunk(chunk: list[str]) -> list[list[float]]:
    result = await embedder.embed_documents(chunk)
    return [list(v) for v in result.embeddings]

vectors = await aembed_in_batches(
    texts=texts, batch_size=96, aembed_chunk=_embed_chunk
)  # result[i] maps to texts[i]; no chunk ever exceeds batch_size

The helper is embedder-agnostic (it takes an async chunk-embedder callable), so this package carries no pydantic-ai/vendor dependency. Every list-embed call site (chat memory store, GraphRAG entity index, RAG ingest) routes through it.

Version 1.24.0 Subagent Primitive Contracts

• Adds framework-neutral subagent schema contracts in keble_helpers.ai.subagent: archetype kind/mode/outcome enums, escalation policy, run budgets, ToolScope, DelegationBrief, SubAgentArchetypeDescriptor, SubAgentProviderManifest, and SubAgentProviderProtocol.
• DelegationBrief.user_request_verbatim is the required acceptance anchor for scoped agents; parent-authored paraphrase is no longer the source of truth.
• ChatProviderFamily.SUB_AGENT is the provider-family value for generated delegation/supervision tools. The old BACKGROUND_SESSION family enum value is intentionally removed in the coordinated Round-4 train.

Version 1.23.0 Agentic Evidence Dedupe Identity

• AgenticEvidenceItem.canonical_key() is the shared evidence-artifact identity used by chat runtimes to dedupe chips. It includes kind, url, and sorted ref items, and intentionally excludes label because labels are display copy.
• Relabeling the same report/product/file/link should not create another timeline chip. Changing the domain reference should create a distinct chip.

Version 1.12.16 Aliyun Python 3.13 import compatibility

• Preserves the existing from keble_helpers import AliyunOss public export.
• Adds explicit Python 3.13 compatibility dependencies for Aliyun OSS imports: legacy-cgi, six, and the aliyun-python-sdk-core 2.16.x line.
• Bridges the Aliyun SDK's older vendored six.moves module paths at module import time, so downstream packages can import helper schemas without install-order-dependent failures.

Version 1.12.11 update

• Adds package-neutral UsageAccountingEvent, UsageAccountingRecorderProtocol, UsageAccountingSource, and UsageAccountingUnitType.
• The usage-accounting contract intentionally contains no MongoDB, task id, pricing, or backend business logic. Host services decide how to price and persist emitted events.
• Counted API/item/search events require unit_count; token events require Pydantic-AI RunUsage.

Shared Typings

keble-helpers owns package-neutral shared enums and value objects that need to stay stable across backend packages.

1. Marketplace
2. Language
3. CommerceEntityType

CommerceEntityType is the canonical cross-package enum for commerce entities:

1. BRAND_IDENTITY
2. BRAND_MENTION
3. CATEGORY
4. PRODUCT
5. LISTING
6. SKU

Display helpers on CommerceEntityType are part of the shared contract as well. upper_snake_to_title() must preserve known acronyms such as SKU instead of degrading them to title-cased words.

Version 1.12.10 update

• Versions the helper publication-blocker documentation on the maintained 1.12.x line.
• Runtime helper schemas and protocols are unchanged from 1.12.9.

Agent Runtime Context

keble-helpers 1.12.10 includes AgentBaseDeps, a package-neutral Pydantic deps base for browser-selected agent context.

1. marketplace carries the commerce marketplace scope when tools need it.
2. language carries the frontend language that prompts and user-visible agent output should respect.
3. Agent packages can combine it with database deps through multiple inheritance, for example class MyDeps(AgentDbDeps, AgentBaseDeps): ....

Do not expose internal datasource or provider names in user-facing bootstrap chat copy. Keep such names in provider schemas, internal logs, or tests only.

AgentBaseDeps Publication Status

AgentBaseDeps is built in keble-helpers 1.12.10, but this environment cannot publish it to the configured package index unless PyPI credentials or OIDC trusted-publishing token are available.

1. uv build should produce dist/keble_helpers-1.12.10-py3-none-any.whl.
2. python -m pip index versions keble-helpers still shows 1.12.1 as the latest visible index release.
3. If uv publish dist/keble_helpers-1.12.10-py3-none-any.whl fails with missing credentials, service repos must consume the bundled wheel until a credentialed publish is performed.
4. The 2026-05-23 local publish attempt failed for that credential reason, so this workspace still treats 1.12.10 as a built, bundled-wheel release.

ObjectId Boundary Rule

keble_helpers.ObjectId is a Pydantic annotation over BSON ObjectId, not a replacement for every Mongo query id. It serializes to a string when schemas are dumped with model_dump(mode="json").

Use keble_helpers.ObjectId in Pydantic schemas that cross API, chat, tool, or queue boundaries. Use direct BSON ids in Mongo query internals and tests when that is clearer, preferably aliased as BsonObjectId.

Queued Envelope Contract

keble-helpers owns queue-neutral envelope types that let backend and package clients share one broker contract without sharing Celery internals.

1. QueuedEnvelope is the persisted message body:
   • job_type tells processors whether they own the work,
   • payload is opaque to the dispatcher and validated by the handling processor,
   • domain_refs links optional domain objects such as task, grid, or positioning ids.
2. QueuedEnvelopeProcessContext wraps one envelope plus attempt/runtime metadata.
3. QueuedEnvelopeProcessResult tells the backend dispatcher whether a processor handled the envelope and whether handling succeeded.
4. QueuedEnvelopeProcessorProtocol is the shared client-side contract:

async def aprocess_queued_envelope(
    context: QueuedEnvelopeProcessContext,
) -> QueuedEnvelopeProcessResult:
    """Process owned job types and return handled=False for unrelated jobs."""

Backend owns Celery, queue ordering, retries, and persisted ledger status. Package clients own their job-type payload schemas and validation.

Agentic Action Events

keble-helpers owns the package-neutral event envelope used by task, positioning, segmenting, and backend action runtimes.

1. AgenticActionEvent[T] wraps one package-owned typed result payload.
2. AgenticActionEventSource is the single source of truth for the emitting package identity; source is an enum, not a free string, and keble-core mirrors these values 1:1 for the frontend.
3. AgenticActionEventStatus records lifecycle state such as SUCCEEDED or FAILED. It is now a backward-compatible alias of the unified AgenticActionStatus (see "Unified Agentic Action Contract" below).
4. AgenticActionEvent.started/progressed/succeeded/failed(...) are the canonical cls factories. Use them instead of hand-building the envelope so subclasses never inherit a wrong status default.
5. AgenticEventEmitter is non-blocking: aemit schedules one background task wrapping the event's ordered callback chain and returns immediately, so producers never block their data path (or a UI stream) on callback I/O. Callbacks for a single event still run in registration order.
6. AgenticEventEmitter.adrain() awaits all outstanding emit tasks at a workflow boundary. Failures are best-effort, observed — they never fail the producing workflow; each non-cancelled exception is routed to an injected EmissionErrorHook (via bind_error_hook, e.g. backend Sentry) as a package-neutral EmissionError, or logged when no hook is bound. Drain loops to a fixpoint so re-entrant emits are joined.
7. AgenticEventEmitter.build(...) lets callers pass None, callback sequences, or a ready emitter; an existing emitter is returned unchanged so its bound hook and in-flight task registry survive.

event = AgenticActionEvent[MyActionResult].succeeded(
    source=AgenticActionEventSource.KEBLE_SEGMENTING,
    action_type="UPDATE_DIMENSIONS",
    payload=result,
    root_id=str(grid_id),
)
emitter.bind_error_hook(sentry_emission_error_hook)  # backend injects the sink
await emitter.aemit(event)   # non-blocking: schedules + returns
# ... finish the round's durable work ...
await emitter.adrain()       # await all callbacks; surface failures to the hook

Contract: every consumer MUST reach an adrain() boundary before reading a callback side effect (e.g. keble-task spawn-on-terminal, terminal-subtree settle) and before its unit of work returns; otherwise detached emit tasks leak past the boundary.

The helper contract deliberately does not depend on Celery, FastAPI, or SSE. Backend can later bridge the same event envelope into listeners without each package inventing a new callback shape. Every agentic package (task, segmenting, positioning, amz-product-report) sets source from AgenticActionEventSource so the backend Redis publisher and the single task-room WebSocket carry one consistent envelope.

Unified Agentic Action Contract

Raising a browser client tool, asking for a server-tool approval, and self-serving a subagent decision are all the same kind of thing — an action that pauses or reports on an agent run. They therefore share one canonical contract in keble_helpers/ai/client_action.py, instead of each host re-defining its own status/progress/kind enums (the former keble_agentic_chat.ChatActionStatus / ChatActionProgress / ChatActionKind are now removed in favor of these).

1. AgenticActionStatus is the single superset lifecycle enum. It carries both the event-lifecycle members (STARTED/PROGRESSED/SUCCEEDED/FAILED) and the resolution members (PENDING/APPROVED/DENIED/SUBMITTED/REJECTED/ABANDONED). AgenticActionEventStatus is now a backward-compatible alias of this one enum, so event runtimes and action runtimes never disagree on a status string.
2. AgenticActionKind enumerates the four families: SERVER_TOOL_APPROVAL, CLIENT_TOOL, SERVER_SELF_SERVED (a background subagent answering its own client tool with the most-likely user choice, no human in the loop), and SERVER_TOOL_CALL (an auto-executed backend tool call recorded for the visible timeline — created directly terminal SUCCEEDED/FAILED, never pending, never resolvable).
3. AgenticActionProgress is the canonical progress shape (the renamed home of the former ChatActionProgress), with .queued(...) and .to_action_status().
4. AgenticClientActionBase is the package-neutral base every host action inherits (keble_agentic_chat.ChatAction inherits it). It owns identity (action_id, tool_call_id, tool_name), kind, optional source (AgenticActionEventSource), status, JSON-safe request/result, optional progress, and timestamps.

Rule for all packages: any action that pauses or reports an agent run inherits AgenticClientActionBase and uses AgenticActionStatus/AgenticActionProgress. No package re-defines a parallel status or progress enum.

from keble_helpers import (
    AgenticActionKind, AgenticActionStatus, AgenticClientActionBase, AgenticActionEventSource,
)

class ChatAction(AgenticClientActionBase, ChatValueBase):
    """A host action inherits the canonical contract and only adds serialization policy."""

action = ChatAction(
    action_id=tool_call_id, kind=AgenticActionKind.CLIENT_TOOL,
    source=AgenticActionEventSource.KEBLE_AGENTIC_CHAT,
    tool_call_id=tool_call_id, tool_name="request_client_tool",
    status=AgenticActionStatus.PENDING, request={"tool_type": "MARKETPLACE_SELECT"},
    occurred_at=datetime.now(timezone.utc),
)

Agentic Chat Scope Runtime (cross-repo standard)

ChatScopeRuntimeProtocol (keble_helpers/ai/chat_scope.py) is the framework-neutral contract for "a per-scope agentic-chat runtime holder" — the object that owns one chat scope's durable history store plus the cooperative run-control used to stop an in-flight streaming run. The concrete implementation is keble_agentic_chat.AgenticChat (it binds a LangGraph turn engine and adds a per-turn runtime factory). The contract is declared HERE, in framework-neutral keble-helpers (no LangGraph / pydantic-ai types — store and chat_id are Any), so any consumer package can type against it without adding keble-agentic-chat as a dependency.

Standard for future agents: when a package or service needs to host an agentic-chat surface, type against ChatScopeRuntimeProtocol and reuse the package AgenticChat instead of re-inventing store/interrupt ownership inside a service repo.

from keble_helpers import ChatScopeRuntimeProtocol

async def stop_run(scope: ChatScopeRuntimeProtocol, *, owner: str, scope_id: str, chat_id: str) -> bool:
    """Type against the neutral contract; no keble-agentic-chat import needed."""
    return await scope.arequest_interrupt(
        owner=owner, scope_type="TASK", scope_id=scope_id, chat_id=chat_id, reason="user requested stop",
    )

Chat tool providers (cross-repo standard)

ChatToolProviderProtocol (keble_helpers/ai/chat_tool_provider.py, since 1.15.0; contract tightened in 1.32.0) is the companion contract for "a domain that contributes tools to a chat scope". Instead of a host hardcoding every domain's register_*_tools call inline in a giant per-scope builder, each domain is wrapped as a small provider that exposes three members:

1. provider_id — a ChatToolProviderId enum member (the closed, cross-repo set of diagnostics ids; keble-core mirrors it 1:1 in TypeScript). NOT a bare string — the value is persisted and frontend-labelled, so it must be typed.
2. manifest — its ChatToolProviderManifest (first-class as of 1.32.0; manifest.provider_id must equal provider_id).
3. register(*, agent) — attach the tools. There is no context parameter: every provider captures its per-request deps at construction time (a runtime context was always dead — every host passed None and every implementation discarded it).

The host composes a list of them with keble_agentic_chat.compose_tool_providers, which asserts manifest↔registration parity, rejects duplicate ids, and verifies manifest.provider_id is provider_id. The contract is framework-neutral (agent is Any), so a provider adapter need not import keble-agentic-chat or pydantic-ai.

Standard for future agents: to give a chat scope a new domain's tools, add a provider that satisfies ChatToolProviderProtocol and append it to the scope's provider list — do not edit the scope builder's internals or genericize the per-scope deps. To add a brand-new provider id, add ONE member to ChatToolProviderId (and mirror it in keble-core + a frontend lang key).

from typing import Any

from keble_helpers import (
    ChatToolKind,
    ChatToolProviderId,
    ChatToolProviderManifest,
    ChatToolProviderProtocol,
    ChatToolSpec,
)


class MyDomainToolProvider:
    """Wrap an existing `register_my_tools` registrar as a chat tool provider."""

    provider_id = ChatToolProviderId.MEMORY  # an existing enum member

    def __init__(self, *, client: Any) -> None:
        self._client = client

    @property
    def manifest(self) -> ChatToolProviderManifest:
        return ChatToolProviderManifest(
            provider_id=self.provider_id,
            tools=[ChatToolSpec(name="my_tool", kind=ChatToolKind.QUERY, summary="…")],
        )

    def register(self, *, agent: Any) -> None:
        register_my_tools(agent, client=self._client)  # existing domain registrar


_: ChatToolProviderProtocol = MyDomainToolProvider(client=object())

Chat memory contracts (cross-repo standard)

ChatMemoryRecord, ChatMemoryKind, and ChatMemoryStoreProtocol (keble_helpers/ai/chat_memory.py, since 1.16.0) are the framework-neutral durable-memory seam for agentic chats. The chat engine (keble_agentic_chat.LangGraphChatRuntime) recalls records before a turn and remembers new ones after it, but the record shape and store protocol live here so domain packages and hosts produce/consume the same records without importing the engine. kind stays a plain str for host-defined kinds; well-known values (EPISODE/FACT/PREFERENCE/SUMMARY) come from ChatMemoryKind, and the per-turn episodic record is built via ChatMemoryRecord.episode(...).

Scoping contract (locked by design): recall is shared across ALL chats of the same (owner, scope_type) pair — durable memory intentionally crosses chat sessions. scope_id/chat_id are write-side provenance metadata a store persists for diagnostics, NOT recall filters.

Generic agentic memory (since 1.25.0, additive — every existing call site and the 7 backend chat-memory tests stay valid): ChatMemoryRecord now optionally carries owner (a payload-level owner, e.g. a host SHARED-owner sentinel), links: list[MemoryLink] (typed provenance edges — MemoryLink{kind, ref_id, role} with MemoryLinkKind REPORT|MEMORY|OTHER and MemoryLinkRole FINAL|SUBMARKET|WRONG_CONFIG), and parent_memory_id (nested submarket trail). ChatMemoryStoreProtocol.arecall gains keyword-only DEFAULTED filters kinds, since, until, and include_shared=False; chat recall passes none of them and behaves exactly as before, while discovery/niche recall and the agentic search_memories tool pass include_shared=True (plus a time window and kinds) to union the shared-owner namespace. The new aupdate(*, owner, scope_type, memory_id, text=None, metadata=None) -> bool is the mutation half of the agentic update_memory tool (owner+scope gated, re-embeds on text change).

from keble_helpers import ChatMemoryKind, ChatMemoryRecord, ChatMemoryStoreProtocol


class MyVectorMemoryStore:
    """Host store over any backend (Qdrant, SQL, files...)."""

    async def arecall(self, *, owner, scope_type, scope_id, chat_id, query, limit=8):
        ...  # filter by owner + scope_type only (cross-chat recall by design)

    async def aremember(self, *, owner, scope_type, scope_id, chat_id, records):
        ...  # persist records; scope_id/chat_id stored as provenance metadata


_: ChatMemoryStoreProtocol = MyVectorMemoryStore()

Image Prompt Runtime

keble-helpers 1.12.9 owns the shared image prompt preflight, image-count budget, and provider fallback policy used by backend AI clients.

1. ImagePromptChecker accepts only model-supported image responses: image/png, image/jpeg, image/gif, and image/webp.
2. HTTP probes reject non-200/206 responses, unsupported response Content-Type values such as image/svg+xml, and extension-only URLs outside .png, .jpg, .jpeg, .gif, or .webp.
3. ImagePromptChecker.max_images_per_prompt keeps typed ImageUrl parts below the model/provider limit before the call. Non-image prompt parts are always preserved.
4. arun_with_image_fallback(...) now treats provider invalid_image_format, unsupported-image, and too-many-images errors like inaccessible image URLs: it retries once with image parts stripped, then preserves the original exception if the text-only retry still fails.
5. Backend services should keep using the shared checker instead of adding service-local image validation.
6. arealize_image_prompt_urls(prompt, *, checker=None) (and ImagePromptChecker.arealize_prompt_images) is the PRIMARY defense: it prefetches each ImageUrl to bytes in-memory (aiohttp + tenacity, loop-local semaphore, image-count budget) and swaps it for BinaryContent(data, media_type), so providers never download URLs server-side. A slow/blocked CDN URL otherwise raises a fatal ModelHTTPError 400 ("Timed out while downloading image ..."). Unfetchable images are dropped (degrade), never raised. Multimodal call sites should realize the prompt BEFORE the model call; media_type comes from the response Content-Type, else the URL extension, else image/jpeg.

Agentic Tool Config

keble-helpers owns the shared pydantic-ai tool registration config used by package registrars. Packages should import AgentToolConfig rather than redefining local tool-name/description/approval schemas.

from keble_helpers import AgentToolConfig

config = AgentToolConfig.build(
    {
        "name": "mutate_segmenting",
        "description": "Apply one typed segmenting action batch.",
        "requires_approval": True,
    }
)

This helper only describes tool metadata. Package registrars still own the domain payload type and tool execution body.

Agentic Tool Arguments (*_or_model_retry)

keble-helpers owns the canonical parser for model-supplied ids inside pydantic-ai tools. Every agent query tool MUST use these instead of reimplementing ObjectId.is_valid inline. The key rule: HTTP (ClientSideInvalidParams) and worker (ServerSideInvalidParams) exceptions are correct for their boundaries but are meaningless to an agent — at a tool boundary a bad/missing id must surface as ModelRetry so the model self-corrects.

from keble_helpers import (
    parse_object_id_or_model_retry,
    require_object_or_model_retry,
)

# inside an @agent.tool
oid = parse_object_id_or_model_retry(positioning_id, field_name="positioning_id")
loaded = await client.aget(oid)                      # DB load stays in the tool
positioning = require_object_or_model_retry(
    loaded, field_name="positioning_id", raw=positioning_id
)

Two retryable failure modes, with deliberately distinct messages:

1. parse_object_id_or_model_retry — the raw string is not a valid ObjectId.
2. require_object_or_model_retry — the id was well-formed but nothing loaded (hallucinated-but-valid id, deleted object, or not owned by the caller).

Aliyun

The Aliyun module provides helpers for interacting with Alibaba Cloud (Aliyun) services.

Base Classes

Aliyun

• __init__(*, access_key: str, secret: str): Initialize with Aliyun credentials

OSS (Object Storage Service)

AliyunOss

• __init__(oss_endpoint: AnyHttpUrl, bucket: str, **kwargs): Initialize OSS client
• get_bucket() -> oss2.Bucket: Get OSS bucket instance
• get_bucket_with_sts(sts_token: str): Get bucket with STS token
• get_object_meta(key: str) -> AliyunOssMeta: Get object metadata
• save_object_to_local(key: str, local_path: str, *args, **kwargs): Download file from OSS
• save_local_to_cloud(key: str, local_path: str, *args, **kwargs): Upload file to OSS
• save_snapshot_to_local(key: str, local_path: str, seconds: int): Get video snapshot
• cold_archive_object(key: str): Convert object to cold archive storage class
• get_sts_signed_url(sts_token: str, key: str, *, expire_seconds: int = 60, content_type: Optional[str] = None, oss_storage_class: Optional[str] = None) -> str: Generate signed URL with STS

STS (Security Token Service)

AliyunSts

• __init__(region, **kwargs): Initialize STS client
• get_sts(session_name: str, role_arn: str) -> AliyunStsToken: Get STS token

Schemas

AliyunOssPutObjectResponse

• status: int: Response status
• request_id: str: Request ID
• etag: str: ETag
• headers: dict: Response headers

AliyunStsToken

• access_key_secret: str: Access key secret
• security_token: str: Security token
• access_key_id: str: Access key ID

AliyunOssMeta

• etag: Optional[str]: OSS ETag
• content_length: Optional[int]: File size in bytes
• last_modified: Optional[int]: Last modified timestamp
• content_type: Optional[str]: MIME type of the file

Usage Examples

from keble_helpers import AliyunOss

# Initialize Aliyun OSS
oss_client = AliyunOss(
    oss_endpoint="https://oss-cn-beijing.aliyuncs.com",
    bucket="your-bucket-name",
    access_key="your-access-key-id",
    secret="your-access-key-secret"
)

# Upload file to OSS
response = oss_client.save_local_to_cloud(
    key="path/in/oss/file.txt",
    local_path="/local/path/to/file.txt"
)

# Get file metadata
meta = oss_client.get_object_meta("path/in/oss/file.txt")

# Download file from OSS
oss_client.save_object_to_local(
    key="path/in/oss/file.txt",
    local_path="/local/path/to/download.txt"
)

# Get STS token
sts_client = AliyunSts(
    region="cn-beijing",
    access_key="your-access-key-id", 
    secret="your-access-key-secret"
)
sts_token = sts_client.get_sts(
    session_name="session-name",
    role_arn="acs:ram::your-account-id:role/your-role-name"
)

# Generate signed URL with STS token
signed_url = oss_client.get_sts_signed_url(
    sts_token=sts_token.security_token,
    key="path/in/oss/file.txt",
    expire_seconds=3600
)

Progress

The Progress module provides a Redis-based task tracking system to monitor the progress of multi-stage operations.

Base Classes

ProgressHandler

• __init__(redis: Redis): Initialize with Redis connection
• new(*, key: str, model_key: str | None = None) -> ProgressTask: Create a new progress task
• get(*, key: str) -> ProgressReport | None: Retrieve progress report by key

ProgressTask

• __init__(redis: Optional[Redis] = None, key: Optional[str] = None, model_key: Optional[str] = None, root: Optional["ProgressTask"] = None): Initialize a progress task
• new_subtask() -> ProgressTask: Create a subtask under this task
• success(): Mark task as successful
• failure(error: Optional[str] = None): Mark task as failed
• set_message(message: Optional[str]): Set a message for the task
• get_from_redis(redis: Redis, *, key: str) -> Optional["ProgressTask"]: Class method to retrieve a task from Redis
• get_prebuilt_subtasks_model(root: "ProgressTask", redis: Redis, *, model_key: str) -> List["ProgressTask"]: Class method to get prebuilt subtasks

Schemas

ProgressTaskStage

Enum with the following values:

• PENDING: Task is in progress
• SUCCESS: Task completed successfully
• FAILURE: Task failed

ProgressReport

• progress_key: Optional[str]: Key used to store progress in Redis
• progress: float: Completion percentage (0.0 to 1.0)
• is_root_success: bool: Whether the root task is successful
• success: int: Number of successful tasks
• failure: int: Number of failed tasks
• pending: int: Number of pending tasks
• assigned: int: Number of assigned tasks
• total: int: Total number of tasks
• message: Optional[str]: Optional message
• errors: List[str]: List of error messages

Usage Examples

import uuid
from redis import Redis
from keble_helpers import ProgressHandler

# Initialize Redis connection
redis = Redis(host='localhost', port=6379, db=0)

# Create a progress handler
handler = ProgressHandler(redis=redis)

# Create a new progress task
task_key = str(uuid.uuid4())
task = handler.new(key=task_key)

# Create subtasks
subtask1 = task.new_subtask()
subtask2 = task.new_subtask()
subtask3 = task.new_subtask()

# Mark tasks as complete or failed
subtask1.success()
subtask2.failure(error="Something went wrong")
subtask3.success()
task.success()

# Get progress report
report = handler.get(key=task_key)
print(f"Progress: {report.progress * 100}%")
print(f"Success: {report.success}, Failure: {report.failure}, Pending: {report.pending}")

# Using model_key for prebuilt subtasks
model_key = str(uuid.uuid4())
root_task = handler.new(key=str(uuid.uuid4()), model_key=model_key)

# When you create a new task with the same model_key,
# it will have the same number of subtasks
new_task = handler.new(key=str(uuid.uuid4()), model_key=model_key)

Pydantic

The Pydantic module provides helpers and utilities for working with Pydantic models.

Functions

• is_http_url(url: Any) -> bool: Validates if a string is a valid HTTP or HTTPS URL by checking if it has a valid HTTP/HTTPS scheme and netloc

Base Classes

PydanticModelConfig

• default_dict(**kwargs) -> dict: Returns a dictionary with default configuration
• default(**kwargs) -> ConfigDict: Returns a ConfigDict with default configuration

CloudStorageBase

• Base model for cloud storage objects with standardized fields

Enums

CloudStorageType

• AWS_S3: Amazon S3 storage
• ALIYUN_OSS: Alibaba Cloud OSS storage

CloudStorageObjectType

• IMAGE: Image files
• VIDEO: Video files
• EXCEL: Excel spreadsheets
• CSV: CSV files
• OTHER: Other file types
• determine_type(*, mime: str) -> CloudStorageObjectType: Determine type from MIME

Usage Examples

from keble_helpers.pydantic import CloudStorageBase, CloudStorageObjectType, CloudStorageType
from keble_helpers.pydantic.schemas import is_http_url
from pydantic import BaseModel

# Check if a URL is valid HTTP/HTTPS
valid = is_http_url("https://example.com")  # True
valid = is_http_url("ftp://example.com")    # False
valid = is_http_url("example.com")          # False (missing scheme)
valid = is_http_url("http://")              # False (missing netloc)

# Create a custom model with Pydantic configuration
class MyModel(BaseModel):
    model_config = PydanticModelConfig.default()
    # Fields go here

# Create a cloud storage object
storage = CloudStorageBase(
    key="path/to/file.jpg",
    base_url="https://example.com/storage",
    type=CloudStorageType.AWS_S3,
    object_type=CloudStorageObjectType.IMAGE,
    original_file_name="photo.jpg"
)

# Determine object type from MIME
object_type = CloudStorageObjectType.determine_type(mime="image/jpeg")

Common

The Common module provides general utility functions for common tasks.

Functions

String and ID Utilities

• id_generator() -> str: Generate a UUID4 string
• generate_random_string(length: int = 32, *, lower: bool = True, upper: bool = True, digit: bool = True) -> str: Generate a random string
• hash_string(arg: str) -> str: Generate MD5 hash of a string
• inline_string(string: str, max_len: int = 30): Format a string for inline display

Pydantic Helpers

• is_pydantic_field_empty(obj: BaseModel, field: str) -> bool: Check if a field is empty in a Pydantic model

Date and Time

• date_to_datetime(d: date) -> datetime: Convert a date to datetime
• datetime_to_date(d: datetime) -> date: Convert a datetime to date
• utc_now() -> datetime: Canonical timezone-aware UTC "now" for persisted models and ledgers (use this instead of inlining datetime.now(timezone.utc))
• ensure_aware_utc(value: datetime) -> datetime: Normalize a datetime to aware UTC, treating naive values as UTC

List and Collection Operations

• chunked(items: Sequence, size: int) -> list[list]: Split a sequence into consecutive chunks of at most size (canonical batch helper; replaces the removed slice_to_list)
• dedupe_preserve_order(items: Iterable) -> list: De-duplicate hashable items, preserving first-occurrence order
• dedupe_keep_order(items, *, key=None) -> list: De-duplicate by explicit identity (key fn, then .key attr, then identity); handles unhashable/keyed objects
• ensure_list(value, *, name: str) -> list: Return a list or raise a typed ServerSideInvalidParams
• DedupeHelpers: grouped classmethods to dedupe keyed/JSON/string items and merge_money_ranges
• normalize_semantic_text(value: str | None) -> str: Collapse whitespace + casefold for semantic name matching
• maybe_await(value): Await value if awaitable, else return it (uniform sync/async handling)
• get_first_match(items: list, key_fn, value): Find first item in a list matching a criterion

File System Operations

• ensure_has_folder(path: str) -> str: Create a directory if it doesn't exist
• yield_files(folder: str) -> Iterator[str | Path]: Recursively yield files in a directory
• get_files(folder: str) -> List[str | Path]: Get a list of all files in a directory
• zip_dir(folder: Path | str, zip_filepath: Path | str): Zip a directory
• remove_dir(dir: Path | str): Remove a directory

MIME Type Checking

• is_mime_prefix_in(mime, mime_start: List[str]): Check if a MIME type has a specific prefix
• is_mime_image(mime: str): Check if a MIME type is an image
• is_mime_video(mime: str): Check if a MIME type is a video
• is_mime_audio(mime: str): Check if a MIME type is audio
• is_mime_media(mime: str): Check if a MIME type is any media (image, video, audio)
• is_mime_ms_excel(mime: str): Check if a MIME type is MS Excel
• is_mime_csv(mime: str): Check if a MIME type is CSV

Usage Examples

from keble_helpers import (
    id_generator, hash_string, ensure_has_folder, get_files,
    is_mime_image, chunked, utc_now,
)

# Generate a unique ID
unique_id = id_generator()

# Generate a hash of a string
file_hash = hash_string("content to hash")

# Ensure a directory exists
path = ensure_has_folder("/path/to/directory")

# Get all files in a directory
files = get_files("/path/to/directory")

# Check if a MIME type is an image
is_image = is_mime_image("image/jpeg")  # True

# One canonical UTC timestamp
created_at = utc_now()

# Split a sequence into chunks of size 3
chunks = chunked([1, 2, 3, 4, 5, 6, 7], 3)  # [[1, 2, 3], [4, 5, 6], [7]]

DateTime

The DateTime module provides utilities for working with dates and times.

Functions

• days_in_month(year, month): Get the number of days in a specific month

Usage Examples

from keble_helpers.datetime import days_in_month

# Get days in February 2024 (leap year)
days = days_in_month(2024, 2)  # 29

# Get days in February 2023 (non-leap year)
days = days_in_month(2023, 2)  # 28

Enum

The Enum module provides predefined enumerations.

Enums

Environment

• development: Development environment
• test: Test environment
• production: Production environment

Usage Examples

from keble_helpers.enum import Environment

# Use environment enum
current_env = Environment.development

# Check environment
if current_env == Environment.production:
    # Production-specific code
    pass

FastAPI

The FastAPI module provides helpers for working with FastAPI applications, focusing on JSON encoding compatible with Pydantic v2.

Functions

• jsonable_encoder(obj: Any, include: Optional[IncEx] = None, exclude: Optional[IncEx] = None, by_alias: bool = True, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, custom_encoder: Optional[Dict[Any, Callable[[Any], Any]]] = None, sqlalchemy_safe: bool = True) -> Any: Convert a Python object to a JSON-compatible object

Constants

• PYDANTIC_V2: Boolean indicating if Pydantic v2 is in use
• ENCODERS_BY_TYPE: Dictionary mapping Python types to encoder functions

Usage Examples

from keble_helpers.fastapi import jsonable_encoder
from pydantic import BaseModel
from datetime import datetime

class User(BaseModel):
    id: int
    name: str
    created_at: datetime
    updated_at: datetime | None = None

user = User(id=1, name="John Doe", created_at=datetime.now())

# Convert to JSON-compatible dict
json_data = jsonable_encoder(user)

# Convert excluding some fields
json_data = jsonable_encoder(user, exclude={"created_at"})

# Convert with custom encoders
json_data = jsonable_encoder(
    user, 
    custom_encoder={datetime: lambda dt: dt.strftime("%Y-%m-%d")}
)

File

The File module provides utilities for file operations, particularly for downloading files.

Functions

• adownload_file(*, url: str, folder: Path, filename: str) -> Path: Asynchronously download a file from a URL

Usage Examples

import asyncio
from pathlib import Path
from keble_helpers.file import adownload_file

async def download_example():
    # Download a file
    file_path = await adownload_file(
        url="https://example.com/file.pdf",
        folder=Path("/path/to/downloads"),
        filename="document.pdf"
    )
    
    print(f"Downloaded to: {file_path}")

# Run the async function
asyncio.run(download_example())

Multithread (Deprecated)

Note: This module is deprecated. The project now uses async-based approaches instead of multithreading.

The Multithread module provides utilities for thread management and parallel execution.

Classes

ThreadController

• __init__(thread_size: int): Initialize with a maximum number of threads
• create_thread(target: Callable, *, args: Optional[tuple] = None, kwargs: Optional[Dict[str, Any]] = None, thread_owner: Optional[str | int] = None, disable_sema: Optional[bool] = False, join: Optional[bool] = False): Create and start a new thread
• acquire(*, thread_owner: Optional[str | int] = None): Acquire a semaphore
• release(*, thread_owner: Optional[str | int] = None): Release a semaphore
• wait_all_to_finish(): Wait for all threads to complete
• wait_owner_to_finish(thread_owner: str | int): Wait for all threads by a specific owner to complete

Decorators

• threaded(*, sema: Optional[Semaphore] = None, join: Optional[bool] = False): Decorator to run a function in a separate thread

Usage Examples

from keble_helpers import ThreadController, threaded
from threading import Semaphore

# Using ThreadController
controller = ThreadController(thread_size=5)

def task(results):
    # Perform task
    results.append("Task completed")
    controller.release()

results = []
for _ in range(10):
    controller.create_thread(target=task, args=(results,))

controller.wait_all_to_finish()

# Using threaded decorator
sema = Semaphore(3)

@threaded(sema=sema)
def background_task(results):
    results.append("Background task completed")
    sema.release()

threads = []
results = []
for _ in range(5):
    threads.append(background_task(results))

for thread in threads:
    thread.join()

NumPy Utils

The NumPy Utils module provides helper functions for working with NumPy arrays and handling numerical values.

Functions

• is_invalid_float(value: Optional[float]) -> bool: Check if a float value is NaN or infinity
• guard_invalid_float(value: float | None | np.floating) -> float | None: Replace invalid float values (NaN, inf) with None

Usage Examples

import numpy as np
from keble_helpers.np_utils import is_invalid_float, guard_invalid_float

# Check if a value is an invalid float
invalid = is_invalid_float(float('nan'))  # True
invalid = is_invalid_float(float('inf'))  # True
invalid = is_invalid_float(42.0)  # False

# Guard against invalid floats
safe_value = guard_invalid_float(np.nan)  # None
safe_value = guard_invalid_float(np.inf)  # None
safe_value = guard_invalid_float(42.0)  # 42.0
safe_value = guard_invalid_float(np.float32(3.14))  # 3.14

Pydantic AI image runtime

keble_helpers.ai.image_runtime provides reusable multimodal runtime hardening helpers for any pydantic-ai callsite that may include typed ImageUrl parts.

Key exports:

• ImagePromptChecker: bounded-concurrency URL preflight with TTL cache and provider-safe image-count budgeting
• ImagePreflightDecision, ImagePreflightBatchResult, ImagePreflightReason: typed preflight decisions
• extract_image_urls(...), strip_image_url_parts(...)
• is_image_url_model_404_error(...)
• arun_with_image_fallback(...): preflight + image-404 text-only fallback retry (tenacity with reraise=True)
• typed_tool(...), typed_tool_plain(...): additive wrappers around agent.tool(...) / agent.tool_plain(...) that preserve the decorated callable type for downstream code

Ownership model:

• host applications should instantiate one shared ImagePromptChecker
• downstream libs should accept that instance and reuse it
• retry policy and preflight policy live on the same checker instance
• libs should not construct hidden per-module checker objects with divergent settings

Status policy (ImagePromptChecker):

• only_accepts: list[int] | None
• rejects: list[int] | None
• provide only one of them (ValueError if both are provided)
• when both are omitted, default behavior is only_accepts=[200, 206]
• probe requests do not auto-follow redirects; 3xx status codes are surfaced to policy evaluation directly

Usage:

from keble_helpers import ImagePromptChecker, arun_with_image_fallback

checker = ImagePromptChecker(
    enabled=True,
    timeout_secs=2.0,
    max_concurrency=4,
    cache_ttl_secs=600,
    max_images_per_prompt=45,
    image_model_404_retry_attempts=1,
    only_accepts=[200, 206],  # strict image status policy
)

result = await arun_with_image_fallback(
    agent=agent,
    prompt=prompt_parts,  # Sequence[UserContent] with optional ImageUrl
    image_prompt_checker=checker,
)
output = result.output  # preserves the agent's concrete output type

# or call through the checker directly
result = await checker.arun_with_image_fallback(
    agent=agent,
    prompt=prompt_parts,
)

Typed tool registration:

from pydantic_ai import Agent
from pydantic_ai.tools import RunContext
from keble_helpers import typed_tool, typed_tool_plain

agent = Agent("test", deps_type=int, output_type=str)

@typed_tool(agent, require_parameter_descriptions=True)
async def repeat(ctx: RunContext[int], word: str) -> str:
    """Repeat a word.

    Args:
        word: Word to repeat.
    """

    return f"{ctx.deps}:{word}"


@typed_tool_plain(agent, name="slugify")
def slugify(name: str) -> str:
    return name.strip().lower().replace(" ", "-")

Contract:

1. Keep domain models/validators pure (no network I/O in validators).
2. Apply runtime preflight right before model invocation.
3. Keep retry ownership on the injected checker instead of separate function kwargs.
4. Preserve terminal exception surfaces (reraise=True) for upstream task/error layers.

Usage Accounting Events

keble_helpers.ai.usage_accounting defines package-neutral usage facts only. Packages should build events with UsageAccountingEvent.from_run_usage(...), from_request_usage(...), or from_unit_count(...) and pass them to an injected UsageAccountingRecorderProtocol.

This helper layer must not import Mongo, keble-task, pricing catalogs, or backend business logic. Sources and units are uppercase Enums so downstream packages and backend recorders share one typed contract.

Use UsageAccountingEvent.from_result_usage(...) when a package records a Pydantic-AI agent result. The classmethod owns normalization for both result.usage value-style APIs and result.usage() method-style APIs, keeping runtime compatibility out of downstream business logic.

UsageAccountingMetadata is the input contract for metadata= and is a covariant Mapping[str, UsageAccountingMetadataValue] (not an invariant dict), so a caller-typed dict[str, str] variable assigns without a cast. The value stored on UsageAccountingEvent.metadata is always a concrete dict[str, UsageAccountingMetadataValue], normalized through the single UsageAccountingEvent.build_metadata(...) helper (None -> {}, else dict(...)). Never re-narrow the input alias back to dict — that reintroduces the value-type invariance error at every consuming call site.