Documentation

APIShift Docs

Same-provider API key pooling for LLM apps with adaptive key rotation, cooldown-aware retries, and optional cross-provider fallback. Available for Python and TypeScript/Node.js.

Python

pip install APIShift

TypeScript

npm install @apishift/core

Core concepts: APIShift treats multiple keys/accounts for the same provider as a pool. It rotates healthy keys, cools down only the limited key, and shifts providers only when the primary pool is unavailable. Context is preserved.

Python Quick Start

Install from PyPI and set up a same-provider key pool in under 10 lines.

terminal

$ pip install APIShift

basic usage

from APIShift import Conversation

conversation = Conversation.from_gemini_key_pool(
    api_keys=['key_1', 'key_2', 'key_3'],
    key_strategy='adaptive',
    system_prompt="You are a helpful assistant.",
)

response = conversation.send_message("Explain quantum computing.")
print(response)

fallback providers shortcut

# Same provider first, fallback providers second
conversation = Conversation.from_free_providers(
    gemini_keys=['key_1', 'key_2'],
    openrouter_keys=['key_a'],
    discover_openrouter_models=True,
    groq_keys=['key_g'],
    memory_path='.apishift/memory.jsonl',
    max_context_tokens=6000,
)

Providers

All providers extend LLMProvider and expose a normalized chat interface with per-key rotation and cooldown tracking.

Provider	Default Model	Priority	Free
GeminiProvider	gemini-1.5-flash	10	Yes
OpenRouterProvider	llama-4-scout:free	20	Yes
GroqProvider	llama-4-scout	30	Yes

Key rotation:Each provider holds multiple API keys. When a key is rate-limited, it's cooled down for the duration specified by the Retry-After header (or key_cooldown_seconds, default 60s). The next healthy key is selected automatically.

Provider sorting: Free-tier providers are always preferred. Within the same tier, lower priority values are tried first.

OpenRouter discovery: Set discover_openrouter_models=True to query the OpenRouter API for all currently zero-priced models at runtime instead of using hardcoded defaults.

Streaming

send_message_streamreturns an iterator that yields string chunks. Failover happens before the first chunk is yielded ("peeking" strategy).

streaming

for chunk in conversation.send_message_stream("Tell me a story."):
    print(chunk, end="", flush=True)

How streaming failover works: 1. APIShift initiates the stream with the current provider 2. Attempts to retrieve the first chunk (the "peek") 3. If the peek fails (429/quota), cools down the key, tries another key in the same pool, then falls back if needed 4. Once the first chunk succeeds, it's yielded to the caller 5. If the stream breaks after yielding, the error is raised — partial data has been consumed

stream with error handling

try:
    for chunk in conversation.send_message_stream("Write a long essay."):
        print(chunk, end="")
except Exception as e:
    print(f"\\nStream interrupted: {e}")

RAG & Persistent Memory

APIShift optionally supports FAISS-backed retrieval and JSONL-based persistent memory.

faiss rag

# Optional: requires faiss-cpu + sentence-transformers
conversation.add_to_faiss("The secret password is 'Swordfish'")

# RAG context is automatically injected into prompt budget
for chunk in conversation.send_message_stream("What is the secret password?"):
    print(chunk, end="")

persistent memory

# JsonMemoryStore: append-only JSONL backend
conversation = Conversation(
    providers,
    memory_path='.apishift/memory.jsonl',
)

# History and summaries survive process restarts
# Older messages are summarized and trimmed automatically

Context Packing

Before each provider call, APIShift compiles messages into a token budget using pack_messages_to_token_budget.

Context compilation order: 1. system_prompt — your static instructions 2. summary — condensed older conversation history 3. retrieved context — FAISS results for the latest query 4. recent messages — newest turns, packed newest-first Token estimation: ~1 token per 4 characters (fast, dependency-free). If system context exceeds the budget, it's truncated from the left (preserving the most recent summary/retrieval text).

inspect compiled context

# Inspect what will be sent to the next provider call
compiled = conversation.get_compiled_context()
for msg in compiled:
    print(f"{msg['role']}: {msg['content'][:80]}...")

Exceptions

All exceptions inherit from MultiLLMManagerError.

Exception	When	Attributes
RateLimitError	HTTP 429 or rate-limit phrases detected	provider, retry_after_seconds, status_code, headers
QuotaExceededError	Daily/monthly quota exhausted	provider, retry_after_seconds, status_code, headers
NoAvailableProvidersError	All providers/keys exhausted	message
ProviderInitializationError	Missing SDK or empty key list	provider, message

Python API Reference

Complete constructor signatures and method reference.

Conversation.init

providers	List[LLMProvider]	[]	Ordered list of providers
max_history_length	int	20	Max recent messages kept
system_prompt	str \| None	None	Static system instructions
key_cooldown_seconds	float	60.0	Default cooldown per key
key_strategy	str	'adaptive'	adaptive, round_robin, or sticky
rag_top_k	int	3	FAISS results per query
enable_rag	bool	True	Enable FAISS indexing
summary_max_chars	int	4000	Max summary length
memory_path	str \| None	None	JSONL persistence path
max_context_tokens	int \| None	6000	Token budget per call

Methods

→
from_gemini_key_pool(api_keys, **kwargs)
Create the primary same-provider Gemini key-pool setup.
→
send_message(message, **kwargs) → str
Send a message and get a full response. Key rotation and fallback are automatic.
→
send_message_stream(message, **kwargs) → Iterator[str]
Stream a response. Pre-yield failover on rate limits.
→
add_to_faiss(message)
Index text into the optional FAISS store for retrieval.
→
retrieve_context(query, top_k) → List[str]
Retrieve semantically relevant stored snippets.
→
get_history() → List[Dict]
Return a copy of the conversation history.
→
get_compiled_context() → List[Dict]
Inspect the exact messages that will be sent on the next call.
→
clear_history()
Clear history, summary, FAISS index, and memory store.
→
add_provider(provider)
Add a provider at runtime. Re-sorts by free/priority.

TypeScript Quick Start

@apishift/core integrates with the Vercel AI SDK. Pass any LanguageModelV1 instance.

terminal

$ npm install @apishift/core ai @ai-sdk/google @ai-sdk/openai

basic setup

import { APIShift } from '@apishift/core';
import { google } from '@ai-sdk/google';

const orchestrator = new APIShift([
  { provider: 'gemini', keyIndex: 0, name: 'gemini-key-1', model: google('gemini-1.5-flash', { apiKey: process.env.GEMINI_KEY_1 }) },
  { provider: 'gemini', keyIndex: 1, name: 'gemini-key-2', model: google('gemini-1.5-flash', { apiKey: process.env.GEMINI_KEY_2 }) },
], {
  keyStrategy: 'adaptive',
  systemPrompt: 'Continue the same task across key changes.',
});

APIShift Class (TS)

The orchestrator wraps generateText and streamText from the Vercel AI SDK, adding same-provider key pools, cooldowns, fallback, and context management.

Constructor: new APIShift(models, options?)

models	Array<LanguageModel \| ModelEntry>	Required	Model instances with optional routing metadata
systemPrompt	string	''	Static system instructions
maxHistoryLength	number	20	Max recent messages kept
summaryMaxChars	number	4000	Max summary length
defaultCooldownMs	number	60000	Default cooldown per model/key entry
keyStrategy	string	'adaptive'	adaptive, round_robin, or sticky
routingStrategy	string	'same_provider_first'	Try all healthy keys in a provider pool before fallback
maxContextTokens	number	6000	Token budget per call
memoryStore	PersistentMemoryStore	undefined	JSONL or custom persistence

TypeScript Streaming

Use streamMessage for context-preserving streaming, or streamText for stateless failover.

streaming

const { textStream } = await orchestrator.streamMessage(
  'Explain the event loop in Node.js'
);

for await (const chunk of textStream) {
  process.stdout.write(chunk);
}

TypeScript Memory

Use JsonMemoryStore for JSONL-based persistence, or implement the PersistentMemoryStore interface for custom backends.

persistent memory

import { APIShift, JsonMemoryStore } from '@apishift/core';

const orchestrator = new APIShift(models, {
  memoryStore: new JsonMemoryStore('.apishift/memory.jsonl'),
});

PersistentMemoryStore interface: load() → { history, summary } saveMessage(message) saveSummary(summary) clear()

TypeScript API Reference

→
generateText(params) → Promise
Stateless text generation with failover. Omit the model parameter.
→
streamText(params) → Promise
Stateless streaming with failover. First-chunk peeking strategy.
→
sendMessage(content, params?) → Promise
Context-preserving generation. History and summary managed automatically.
→
streamMessage(content, params?) → Promise
Context-preserving streaming. Chunks are auto-saved on stream completion.
→
getHistory() → Promise<ModelMessage[]>
Return the current conversation history.
→
getCompiledContext() → Promise<ModelMessage[]>
Inspect the compiled messages for the next call.
→
clearHistory() → Promise<void>
Clear history, summary, and memory store.
→
lastRoute: RouteInfo | undefined
Info about the last successful route (name, attempts, cooldown).

Architecture

How the same-provider key pool and fallback pipeline works internally.

Request Flow: user message → append to history → index in FAISS (optional) → trim history → summarize overflow → compile context (system + summary + RAG + recent) → pack to token budget → try primary provider pool → 429/503? cool down key → try next key → fallback if pool is blocked → success? yield response → save to history → loop until success or NoAvailableProvidersError Provider Selection: sort by: free_tier (True first) → priority (lower first) skip keys where: cooldown_until > now max attempts: total_key_count across all provider pools

FAQ

?
What happens if all keys are rate-limited?
NoAvailableProvidersError is raised. Keys recover automatically after their cooldown period expires.
?
Does failover work mid-stream?
Only before the first chunk is yielded (pre-yield failover). Once bytes reach the caller, mid-stream errors are raised directly — restarting would duplicate/garble output.
?
Can I add fallback providers at runtime?
Yes. Call conversation.add_provider(provider). It is sorted after the primary pool based on free and priority metadata.
?
How does OpenRouter model discovery work?
APIShift queries the OpenRouter /api/v1/models endpoint for models with zero prompt and completion pricing. Falls back to hardcoded defaults if the network request fails.
?
What's the token counting accuracy?
The default counter uses ~4 chars/token heuristic. It's fast and dependency-free. For exact counts, pass a custom token_counter (e.g., tiktoken) via the constructor.
?
Can I use APIShift without the Vercel AI SDK?
Python: Yes, it's standalone. TypeScript: The orchestrator wraps generateText/streamText from the Vercel AI SDK, so that dependency is required.
?
Is FAISS required?
No. RAG is completely optional. Without faiss-cpu and sentence-transformers, APIShift works identically — just without semantic retrieval.

APIShift Docs

Python Quick Start

Providers

Streaming

RAG & Persistent Memory

Context Packing

Exceptions

Python API Reference

Conversation.__init__

Methods

TypeScript Quick Start

APIShift Class (TS)

Constructor: new APIShift(models, options?)

TypeScript Streaming

TypeScript Memory

TypeScript API Reference

Architecture

FAQ

Conversation.init