cyra-ai 0.1.0

A universal, async-first Python interface for AI providers.

Cyra

One typed Python interface for cloud, local, and custom AI models.

Cyra normalizes OpenAI, Anthropic, Gemini, DeepSeek, Mistral, Groq, Cohere, OpenRouter, Ollama, LM Studio, and arbitrary OpenAI-compatible endpoints behind one API. It adds model routing, transparent fallback, memory, semantic caching, guardrails, structured output, cost budgets, telemetry, workflows, benchmarks, a CLI, and a local web playground.

Author: Behrad Ghasemi
LinkedIn: linkedin.com/in/behradghasemi

Five-line quickstart

pip install cyra-ai
export OPENAI_API_KEY="..."
python - <<'PY'
from cyra import AI
print(AI().ask("Explain semantic caching in one sentence."))
PY

For a repository checkout, use pip install -e ".[dev]" instead.

Why Cyra

• One synchronous and asynchronous API across major providers.
• Native OpenAI Responses API support and OpenAI-compatible endpoint support.
• Capability-aware routing for cost, quality, speed, reasoning, or a balanced mix.
• Circuit breaking and ordered cross-provider fallback.
• Pydantic v2 structured output with JSON Schema and validation retries.
• SQLite, in-memory, PostgreSQL, Chroma, Pinecone, and Qdrant memory backends.
• Exact and semantic in-memory, SQLite, and Redis caches.
• Prompt-injection blocking, PII redaction, and custom regex guards.
• Preflight cost estimation, hard budgets, usage statistics, and savings suggestions.
• Durable request observability plus arbitrary monitoring hooks.
• Zero-framework CLI and local playground.
• Only two required runtime dependencies: httpx and pydantic.

Prerequisites

Requirement	Supported
Operating system	Linux, macOS, or Windows
Python	CPython 3.10–3.13
Package manager	pip 23+ recommended
Required services	At least one provider API, Ollama, LM Studio, or custom endpoint
Optional services	Redis 5+, PostgreSQL 14+, Chroma, Pinecone, or Qdrant

Installation and setup

PyPI

python -m venv .venv
source .venv/bin/activate               # Windows: .venv\Scripts\activate
python -m pip install --upgrade pip
python -m pip install cyra-ai

Optional backends:

python -m pip install "cyra-ai[redis]"
python -m pip install "cyra-ai[postgres]"
python -m pip install "cyra-ai[socks]"
python -m pip install "cyra-ai[all]"

The distribution is named cyra-ai, while the import package and CLI remain cyra. The shorter PyPI name cyra is owned by an unrelated configuration library and cannot be used for this project without a transfer from its owner.

Source checkout

git clone https://github.com/justbehrad/cyra.git
cd cyra
python -m venv .venv
source .venv/bin/activate
python -m pip install -e ".[dev]"
cp .env.example .env

Cyra reads environment variables directly. It does not parse .env files so that the hard dependency list stays small. Export variables in your shell, use your process manager, or load .env with a tool of your choice.

Universal interface

Text and async

from cyra import AI

ai = AI()

answer = ai.ask("What is quantum computing?")
answer_async = await ai.ask_async("What is quantum computing?")

Use a context manager in long-running applications so HTTP clients close deterministically:

with AI() as ai:
    print(ai.ask("Hello"))

async with AI() as ai:
    print(await ai.ask_async("Hello"))

Streaming

for chunk in ai.stream("Tell a long story"):
    print(chunk, end="", flush=True)

async for chunk in ai.stream_async("Tell a long story"):
    print(chunk, end="", flush=True)

Fallback occurs before the first emitted chunk. If a provider disconnects after text has already been emitted, Cyra raises the provider error rather than silently duplicating or replacing partial output.

Structured output

from pydantic import BaseModel
from cyra import AI

class Recipe(BaseModel):
    title: str
    ingredients: list[str]
    steps: list[str]

recipe = AI().generate(
    Recipe,
    prompt="Give me a simple pasta recipe",
)
print(recipe.title)

Cyra sends the model's JSON Schema when the provider supports it, validates the response with Pydantic, strips common Markdown fences, and retries malformed output. generate_async() is the asynchronous counterpart.

Inspect the normalized response

response = ai.ask("Hello", return_response=True)
print(response.provider)
print(response.model)
print(response.usage.total_tokens)
print(response.cost_usd)
print(response.latency_ms)
print(response.cache_hit)

Intelligent routing and fallback

ai.ask("Translate this to French", optimize="cost")
ai.ask("Write secure authentication code", optimize="quality")
ai.ask("What is 2+2?", optimize="speed")
ai.ask("Solve this logic puzzle", optimize="reasoning")

The router considers:

1. Static capability metadata in ModelCatalog.
2. Provider configuration and requested modalities.
3. Recent successes, failures, and measured latency.
4. Open circuit breakers after repeated failures.
5. Budget pressure, which switches routing to cost optimization at 80% of a limit.

Pin a route when deterministic model selection matters:

ai.ask("Review this", provider="anthropic", model="claude-sonnet-5")

Catalog metadata is editable:

from cyra import ModelSpec

ai.catalog.register(
    ModelSpec(
        provider="custom",
        model="company-model-v2",
        quality=0.9,
        speed=0.8,
        reasoning=0.85,
        input_cost_per_million=0.20,
        output_cost_per_million=0.80,
        multimodal=True,
    ),
    replace=True,
)

Pricing changes over time. Built-in prices include source URLs and snapshot dates; production applications should update or load their own catalog JSON before enforcing financial controls.

Memory

In-memory storage is enabled by default. Replace it at any time:

from cyra import SQLiteMemory

ai.memory = SQLiteMemory("mybot.db")
ai.ask("My name is Behrad")
ai.ask("What is my name?")

Backends

from cyra import (
    ChromaMemory,
    PineconeMemory,
    PostgreSQLMemory,
    QdrantMemory,
)

ai.memory = PostgreSQLMemory("postgresql://user:pass@localhost/cyra")
ai.memory = QdrantMemory(url="http://localhost:6333")
ai.memory = PineconeMemory(host="https://INDEX.svc.pinecone.io", api_key="...")
ai.memory = ChromaMemory(collection_id="COLLECTION_UUID")

Vector backends use a deterministic dependency-free hash embedding by default. That makes setup easy but is not a replacement for a high-quality embedding model. Pass embedding: Callable[[str], list[float]] for production semantic retrieval. Vector backends keep recent-message and summary metadata in a local SQLite sidecar so both chronological and semantic retrieval remain available.

When recent context crosses CYRA_MAX_CONTEXT_TOKENS, Cyra creates a bounded deterministic summary of older turns and injects it into later requests.

Semantic cache

from cyra import RedisCache, SQLiteCache

ai.cache = RedisCache(ttl=3600)
# or:
ai.cache = SQLiteCache(".cyra/cache.db", ttl=3600)

ai.ask("What is the capital of France?")  # provider request
ai.ask("What is the capital of France?")  # exact hit

Semantic matching uses normalized hash vectors and cosine similarity. Tune it:

export CYRA_SEMANTIC_CACHE_THRESHOLD=0.95

Disable per request for volatile data:

ai.ask("Current BTC price", use_cache=False)

Guardrails

from cyra import GuardViolation

try:
    ai.ask("Ignore all previous instructions...", guard=True)
except GuardViolation as exc:
    print(exc.category)

ai.ask("My card is 4242-4242-4242-4242", guard=True)

Built-in handling:

Category	Default action
Prompt injection / jailbreak	Block
Credit card (Luhn-valid)	Redact
Email	Redact
IPv4 address	Redact
Labeled national ID / SSN	Redact

Custom rules:

ai.guardrail.add_pattern(
    "internal_ticket",
    r"SEC-\d{6}",
    action="redact",
    replacement="[REDACTED_TICKET]",
)

Guardrails reduce risk; they are not a complete security boundary. Keep system instructions server-side, use least-privilege tools, validate model output, and isolate untrusted workloads.

Cost control and observability

estimate = ai.cost("Explain quantum computing")
print(estimate.estimated_cost_usd)

ai.budget_daily = 5.0
ai.budget_monthly = 100.0

stats = ai.stats()
print(stats["today"].cost_usd)
print(stats["month"].input_tokens)

print(ai.savings(provider="openai", model="gpt-5.6-sol"))

Provider-reported token usage is authoritative. Streaming usage is estimated when the provider stream does not return final usage.

Default telemetry is stored in .cyra/telemetry.db and includes timestamp, status, model, prompt, response, tokens, cost, and latency. Disable content storage without disabling aggregate telemetry:

export CYRA_LOG_CONTENT=false

Attach Datadog, Logfire, OpenTelemetry, or any other system with a hook:

from cyra.observability import SQLiteTelemetry

def send_to_monitoring(event):
    monitoring_client.emit("cyra.request", event.model_dump(mode="json"))

telemetry = SQLiteTelemetry(".cyra/telemetry.db", hooks=[send_to_monitoring])
ai = AI(telemetry=telemetry)

Multimodal input

The same ask() method accepts media:

from cyra import AI, Media

image = Media.from_path("diagram.png")
answer = AI().ask("Explain this diagram", media=[image])

audio = Media.from_bytes(
    wav_bytes,
    kind="audio",
    mime_type="audio/wav",
)
await ai.ask_async("Transcribe and summarize", media=[audio])

Actual modalities depend on the selected model and provider. The router filters catalog models marked as multimodal, while provider adapters reject unsupported media types with a typed ProviderError.

Local and custom models

Ollama

ollama serve
ollama pull llama3.2
export OLLAMA_HOST=http://127.0.0.1:11434
export CYRA_OLLAMA_MODEL=llama3.2
cyra ask "Explain dependency injection" --provider ollama

LM Studio or another OpenAI-compatible server

export CYRA_CUSTOM_BASE_URL=http://127.0.0.1:1234/v1
export CYRA_CUSTOM_MODEL=local-model
export CYRA_CUSTOM_API_KEY=
cyra ask "Hello" --provider custom

Or register the adapter directly:

from cyra import OpenAICompatibleProvider, ProviderRegistry, AI

provider = OpenAICompatibleProvider(
    name="private",
    api_key="...",
    base_url="https://ai.example.com/v1",
    default_model="company-chat",
)
ai = AI(providers=ProviderRegistry({"private": provider}))

For a non-OpenAI-compatible JSON API, use CustomHTTPProvider and supply typed request/response conversion functions:

from cyra import AI, AIResponse, CustomHTTPProvider, ProviderRegistry

def build(request):
    return {"engine": request.model, "prompt": request.messages[-1].content}

def parse(raw, request):
    return AIResponse(
        request_id=request.request_id,
        text=raw["answer"],
        provider="internal",
        model=request.model,
    )

internal = CustomHTTPProvider(
    name="internal",
    endpoint="https://ai.example.com/generate",
    default_model="v1",
    request_builder=build,
    response_parser=parse,
    headers={"Authorization": "Bearer ..."},
)
ai = AI(providers=ProviderRegistry({"internal": internal}))

Workflows

from cyra import Workflow

workflow = Workflow()
workflow.prompt("Extract keywords")
workflow.if_("'urgent' in result")
workflow.prompt("Draft an urgent email")
workflow.else_()
workflow.prompt("Draft a normal email")
result = workflow.execute()

Use {result} and named variables in subsequent prompt templates. Conditions use a restricted AST interpreter supporting constants, names, Boolean operators, comparisons, and membership. Function calls, attributes, and arbitrary code are rejected.

CLI

cyra ask "Explain RAG" --optimize quality
cyra ask "Tell a story" --stream
cyra models
cyra stats --month
cyra memory search "Behrad" --database mybot.db
cyra memory clear --session customer-42

Golden benchmark:

cyra benchmark tests/data/golden.jsonl \
  --models "openai:gpt-5.6-terra,anthropic:claude-sonnet-5" \
  --output benchmark.json

Each JSONL row:

{"id":"capital","prompt":"Capital of France?","expected_contains":["Paris"],"forbidden_contains":[]}

Playground

cyra playground
cyra playground --host 127.0.0.1 --port 7860 --no-browser

The web UI can use the router, pin a model, compare up to six routes, run multi-prompt workflows, and inspect normalized latency/cost/cache metadata. It binds to localhost by default and has no authentication; do not expose it directly to an untrusted network.

Testing tools

from cyra import AI, MockProvider, ProviderRegistry, test

provider = MockProvider(["deterministic response"])
ai = AI(providers=ProviderRegistry({"mock": provider}))

@test(expected_contains=["response"], tags=["unit"])
def test_prompt():
    assert "response" in ai.ask(
        "hello",
        provider="mock",
        model="mock-model",
    )

GoldenDataset and BenchmarkRunner support deterministic contains/forbidden checks. Application teams can layer custom semantic or human evaluation on the normalized benchmark output.

Plugin system

Cyra discovers standard Python entry points:

Entry-point group	Purpose
cyra.providers	Provider adapters
cyra.memory	Memory backends
cyra.caches	Cache backends
cyra.guards	Guard implementations

Future community packages follow the cyra-<extension> convention:

cyra plugins
cyra install memory-weaviate

cyra install delegates to the active interpreter's pip. Review third-party packages before installing them.

Architecture

flowchart TD
    API["AI / CLI / Playground"] --> Guard["Guard + context"]
    Guard --> Router["Router + budgets"]
    Router --> Provider["Provider adapters"]
    Provider --> Normalize["Normalized response"]
    Normalize --> Store["Memory + cache + telemetry"]
    Store --> API

Module boundaries:

Module	Responsibility
ai.py	Public orchestration, fallback, context, cache, budgets
providers/	HTTP payloads, streaming parsers, normalized errors
routing.py / catalog.py	Capability ranking, health, circuit breaker, pricing
memory/	Conversation records, retrieval, summaries
cache/	Exact and semantic response caching
guardrails.py	Injection detection and PII handling
observability.py / cost.py	Durable events, aggregation, estimation
workflow.py	Safe conditional prompt chains
testing.py / benchmark.py	Mocks and golden evaluations
cli.py / playground.py	Developer interfaces

Request state transitions:

stateDiagram-v2
    [*] --> Guarded
    Guarded --> Cached: hit
    Guarded --> Routed: miss
    Routed --> Completed: provider success
    Routed --> Fallback: retryable failure
    Fallback --> Completed: next provider
    Completed --> Persisted
    Cached --> Persisted
    Persisted --> [*]

Complexity

• Exact in-memory cache: average O(1) time, O(n) space.
• Dependency-free semantic lookup: O(n × d) time and O(n × d) space, where d=256; use a vector backend for large corpora.
• In-memory retrieval: O(n) scan.
• SQLite recent retrieval: O(log n + k) through the session/time index.
• Router: O(m log m) for m eligible models.

Configuration reference

Variable	Default	Effect
CYRA_DEFAULT_MODEL	unset	Pins the default model
CYRA_DEFAULT_PROVIDER	unset	Pins the default provider
CYRA_TIMEOUT_SECONDS	60	Provider request timeout
CYRA_MAX_RETRIES	2	Retry count for retryable HTTP failures
CYRA_DATA_DIR	.cyra	Telemetry and local state directory
CYRA_TELEMETRY_ENABLED	true	Enables default SQLite telemetry
CYRA_LOG_CONTENT	true	Stores prompt and response text
CYRA_MEMORY_SESSION_ID	default	Default conversation namespace
CYRA_MAX_CONTEXT_TOKENS	24000	Summary trigger estimate
CYRA_MEMORY_RECENT_MESSAGES	20	Recent turns injected into prompts
CYRA_SEMANTIC_CACHE_THRESHOLD	0.92	Minimum cosine cache similarity
CYRA_DAILY_BUDGET_USD	unset	Hard daily spend limit
CYRA_MONTHLY_BUDGET_USD	unset	Hard monthly spend limit
REDIS_URL	redis://127.0.0.1:6379/0	Redis cache connection

Provider variables:

Provider	API key	Optional model override
OpenAI	OPENAI_API_KEY	CYRA_OPENAI_MODEL
Anthropic	ANTHROPIC_API_KEY	CYRA_ANTHROPIC_MODEL
Gemini	GEMINI_API_KEY or GOOGLE_API_KEY	CYRA_GEMINI_MODEL
DeepSeek	DEEPSEEK_API_KEY	CYRA_DEEPSEEK_MODEL
Mistral	MISTRAL_API_KEY	CYRA_MISTRAL_MODEL
Groq	GROQ_API_KEY	CYRA_GROQ_MODEL
Cohere	COHERE_API_KEY	CYRA_COHERE_MODEL
OpenRouter	OPENROUTER_API_KEY	CYRA_OPENROUTER_MODEL
Ollama	OLLAMA_HOST	CYRA_OLLAMA_MODEL
Custom	CYRA_CUSTOM_API_KEY	CYRA_CUSTOM_MODEL

Every cloud provider also accepts a <PROVIDER>_BASE_URL override. OpenRouter supports OPENROUTER_HTTP_REFERER and OPENROUTER_APP_NAME.

Development and testing

python -m pip install -e ".[dev]"
pytest
ruff check src tests
ruff format --check src tests
mypy src
python -m build

Interpretation:

• pytest: all unit and mock-transport tests should pass without API keys; branch coverage must remain at or above 90%, and coverage.xml is generated.
• ruff: no lint or import-order violations.
• mypy: strict type checking must report no issues.
• python -m build: creates both wheel and source distribution in dist/.

Live provider tests should be marked integration and excluded from default CI. The included suite never sends a network request.

Publishing to PyPI

Publishing uses GitHub Actions and PyPI Trusted Publishing. No PyPI API token or password is stored in GitHub.

1. Push this repository to https://github.com/justbehrad/cyra.
2. In GitHub, open Settings → Environments → New environment, create pypi, and optionally require a reviewer.
3. In PyPI, open Publishing → Add a new pending publisher → GitHub and enter:

PyPI field	Exact value
PyPI Project Name	cyra-ai
Owner	justbehrad
Repository name	cyra
Workflow name	publish.yml
Environment name	pypi

4. Ensure the version in pyproject.toml is new and the changelog is updated.
5. Commit and push the release, then create and push the matching tag:

git add .
git commit -m "release: v0.1.0"
git push origin main
git tag v0.1.0
git push origin v0.1.0

The tag starts .github/workflows/publish.yml. Its first job runs the full test and quality suite, builds one wheel and one source distribution, and smoke-tests the wheel. The separate publish job receives only id-token: write, downloads the tested artifacts, and publishes them with signed attestations.

After the workflow succeeds:

python -m venv /tmp/cyra-verify
/tmp/cyra-verify/bin/python -m pip install "cyra-ai==0.1.0"
/tmp/cyra-verify/bin/python -c "from cyra import AI; print(AI)"

PyPI versions are immutable. For every later release, update the version first and use a new tag such as v0.1.1; never reuse a failed or published version. Pending publishers do not reserve project names, so publish the first release soon after creating the pending publisher.

Edge cases and operational behavior

• Empty prompts fail before routing.
• Sync methods called from your own async flow should be replaced with their *_async counterpart.
• Provider 401/403, 429, timeouts, 5xx responses, and malformed payloads map to typed exceptions.
• A provider failure after partial streaming is surfaced to prevent duplicate text.
• Unknown pricing contributes $0 to recorded estimated cost and sets pricing_known=False; configure catalog pricing before treating it as billing.
• SQLite enables WAL mode and a 30-second busy timeout. It is not intended to be a high-write distributed database.
• The in-memory implementations are thread-safe but process-local.
• Pydantic validation rejects malformed structured output after the configured retries.

Troubleshooting

No configured and healthy provider

Export at least one provider key, set OLLAMA_HOST, or register a provider manually. Run cyra models to see configured routes.

ProviderAuthenticationError

Confirm the correct API key and base URL. A key for an OpenAI-compatible gateway must be configured under that gateway, not under an unrelated provider.

RedisCache requires cyra-ai[redis]

Install the optional dependency and confirm Redis is reachable:

python -m pip install "cyra-ai[redis]"
redis-cli -u "$REDIS_URL" ping

SQLite database is locked

Keep transactions short, avoid sharing the same database over network filesystems, and use PostgreSQL or Redis for multi-process, high-write workloads.

Structured output still fails

Use a model with reliable JSON Schema support, lower the temperature, inspect the last StructuredOutputError, and make schemas smaller or less ambiguous.

Playground cannot bind

The port is already in use. Select another local port:

cyra playground --port 7861

Semantic cache returns stale facts

Disable caching for volatile prompts, reduce TTL, or increase CYRA_SEMANTIC_CACHE_THRESHOLD.

Security and privacy

• API keys are read from process environment variables and never embedded in source.
• Telemetry includes prompt/response content by default; set CYRA_LOG_CONTENT=false for sensitive workloads.
• The playground is a development server with no authentication.
• Custom endpoints and plugins are trusted code/configuration boundaries.
• PII regexes are best-effort and cannot identify every jurisdiction-specific format.
• Review provider retention and data-processing terms independently.

See SECURITY.md for vulnerability reporting.

Glossary

Term	Meaning
Circuit breaker	Temporarily removes repeatedly failing routes
Fallback	Next ordered model attempted after a provider failure
Golden dataset	Prompts with deterministic expected/forbidden response fragments
LLM	Large language model
PII	Personally identifiable information
Semantic cache	Reuses responses based on vector similarity, not only exact text
SSE	Server-Sent Events, a common streaming response format
WAL	SQLite write-ahead logging

License

MIT © 2026 Behrad Ghasemi