Quick Start#

Installation#

Install the base package plus the provider extras you need:

# Base (no provider)
pip install core-genai

# Individual providers
pip install "core-genai[claude]"
pip install "core-genai[chatgpt]"
pip install "core-genai[gemini]"
pip install "core-genai[grok]"

# Multiple providers at once
pip install "core-genai[gemini,claude,chatgpt,grok]"

# Development (includes all test/lint tools)
pip install -e ".[dev]"

Usage#

All agents share the same interface. Pick a provider, create an agent, and call analyze.

Async (recommended)#

import asyncio
from core_genai.interfaces import IAgent

agent = IAgent.create_agent("GeminiAgent", api_key="YOUR_API_KEY")

async def main():
    output = await agent.analyze(
        model="gemini-2.0-flash",
        prompt="Explain how AI works in a few words.",
    )
    print(agent.get_text(output))

asyncio.run(main())

Synchronous#

Wrap any agent with SyncWrapper from core-mixins to use it synchronously without managing an event loop yourself:

from core_mixins.decorators.async_ import SyncWrapper
from core_genai.interfaces import IAgent

agent = IAgent.create_agent("ClaudeAgent", api_key="YOUR_API_KEY")

with SyncWrapper(agent) as sync_agent:
    output = sync_agent.analyze(
        model="claude-haiku-4-5",
        prompt="Explain how AI works in a few words.",
    )
    print(agent.get_text(output))

Switching providers#

Because every agent implements the same IAgent interface you can swap providers by changing a single line:

# Gemini
agent = IAgent.create_agent("GeminiAgent", api_key=GEMINI_KEY)
output = await agent.analyze(model="gemini-2.0-flash", prompt=PROMPT)

# Claude
agent = IAgent.create_agent("ClaudeAgent", api_key=CLAUDE_KEY)
output = await agent.analyze(model="claude-haiku-4-5", prompt=PROMPT)

# ChatGPT
agent = IAgent.create_agent("ChatGPTAgent", api_key=OPENAI_KEY)
output = await agent.analyze(model="gpt-4o-mini", prompt=PROMPT)

# Grok
agent = IAgent.create_agent("GrokAgent", api_key=XAI_KEY)
output = await agent.analyze(model="grok-3-mini", prompt=PROMPT)

# The rest of the code is identical for all providers:
print(agent.get_text(output))
print(agent.get_cost(output))
print(agent.get_metadata(output))

Usage metadata#

get_metadata() returns a normalized UsageMetadata dict regardless of provider:

meta = agent.get_metadata(output)

print(meta["input_tokens"])         # tokens sent
print(meta["output_tokens"])        # tokens received
print(meta["total_tokens"])         # combined
print(meta["cost_usd"])             # estimated USD cost

# Optional fields (present only when the provider reports them):
meta.get("cached_tokens")           # cache read/hit tokens (Gemini, Claude)
meta.get("cache_creation_tokens")   # cache write tokens (Claude only)
meta.get("reasoning_tokens")        # thinking tokens (Gemini only)

Batch jobs#

Gemini, Claude, ChatGPT, and Grok all implement the IScheduler interface for large-scale asynchronous inference at reduced cost (~50% discount on most providers).

Each agent accepts simple ScheduledJobMetadata request objects; the provider-specific JSONL serialization is handled internally.

import asyncio
from core_genai.interfaces import IAgent, IScheduler
from core_genai.chatgpt.agent import BatchRequest

agent = IAgent.create_agent("ChatGPTAgent", api_key="YOUR_API_KEY")

requests = [
    BatchRequest(custom_id="req-0", prompt=[{"role": "user", "content": "What is ML?"}]),
    BatchRequest(custom_id="req-1", prompt=[{"role": "user", "content": "What is a neural net?"}]),
]

async def main():
    job = await agent.schedule_job(requests=requests, model="gpt-4o-mini")
    print(job["job_id"])

    # Poll until done
    while True:
        state = await agent.check_job_status(job["job_id"])
        if state in agent.TERMINAL_STATES:
            break
        await asyncio.sleep(30)

    result = await agent.extract_job_results(job["job_id"])
    for item in result["results"] or []:
        print(item)

asyncio.run(main())

The same pattern works for Gemini, Claude, and Grok by swapping the agent and request type. Results are returned via ScheduledJobResponse with a normalized job_id, results, error, and state field.