Base Interface#

Agent Interface#

class core_genai.interfaces.IAgent(api_key: str, **kwargs)[source]#

Bases: IFactory

Interface class for all agents.

__init__(api_key: str, **kwargs) → None[source]#

classmethod registration_key() → str[source]#: It returns the name (reference) for the key used to register, like: return self.__name__

classmethod create_agent(cls_name: str, api_key: str, **kwargs) → Any[source]#: Instantiates and returns an agent by its registered class name.

property client: Any#: Must return the client object.

abstractmethod async analyze(model: str, prompt: Any, **kwargs) → Any[source]#: Given a model and prompt the concrete agent must do the analysis.

abstractmethod get_text(output: Any) → str[source]#: Given a model output, it retrieves the response (text information).

abstractmethod get_texts(output: Any) → List[str][source]#: Given a model output, it retrieves all completion choices/candidates.

abstractmethod get_cost(output: Any) → float[source]#: Returns the estimated cost in USD for the given response.

abstractmethod get_metadata(output: Any) → UsageMetadata[source]#: Given a model output, it retrieves the metadata (metadata, cost, etc.) of request.

Scheduler Interface#

class core_genai.interfaces.IScheduler[source]#

Bases: ABC

Interface for providers that support batch (scheduled) job execution.

abstractmethod async schedule_job(requests: List[Any], model: str, **kwargs) → Any[source]#: Submits a batch of requests for deferred execution.

abstractmethod async check_job_status(job_name: str) → Any[source]#: Returns the current status of a scheduled job.

abstractmethod async extract_job_results(job_name: str) → ScheduledJobResponse[source]#: Retrieves the results of a completed scheduled job.

Contracts#

class core_genai.interfaces.UsageMetadata[source]#

Normalized token usage returned by every agent’s get_metadata.

input_tokens: int#

output_tokens: int#

total_tokens: int#

cached_tokens: int#: Cache read/hit tokens.

cache_creation_tokens: int#: Cache write tokens. Claude only.

reasoning_tokens: int#: Thinking/reasoning tokens. Gemini and Grok only.

cost_usd: float#: Estimated total cost in USD.

class core_genai.interfaces.ScheduledJobMetadata[source]#

Normalized response contract for scheduling a batch job across all providers.

job_id: str#: Provider-assigned batch job identifier.

model: str#: Model used for the batch requests.

batch_name: NotRequired[str]#: Human-readable batch name. Grok only.

created_at: NotRequired[datetime]#: Timestamp when the batch job was created.

class core_genai.interfaces.ScheduledJobResponse[source]#

Normalized response contract for batch job results across all providers.

job_id: str#: Provider-assigned batch job identifier.

results: List[Any] | None#: List of individual request results; None if the job has not completed successfully.

error: Any | None#: Provider error details; None if the job succeeded or has not yet finished.

state: Any#: Current job state as reported by the provider.

start_time: NotRequired[datetime | None]#: Job start time. Gemini only.

end_time: NotRequired[datetime | None]#: Job end time. Gemini only.

usage: NotRequired[UsageMetadata | None]#: Aggregated token usage and cost across all succeeded requests. None if the job has not reached a terminal state or all requests failed.