π Readme:
A minimal Python SDK to switch between LLM providers in one line.
No frameworks. No servers. No overengineering.
pip install "askai-python[openai,groq]"Provider SDKs are optional dependencies. Install the extras you need:
pip install "askai-python[openai]"for OpenAI, Azure, and OpenRouter.pip install "askai-python[anthropic]",pip install "askai-python[google]", orpip install "askai-python[groq]"for those providers.pip install "askai-python[all]"for every provider SDK.
Python 3.10 or newer is required.
from ask_ai import OpenAI, Groq
# Auto-detects OPENAI_API_KEY from environment
OpenAI().ask("Explain black holes like I'm 5").text
# Switch provider effortlessly
Groq().ask("Explain black holes like I'm 5").text- One function: Just call
.ask() - Multiple providers: OpenAI, Anthropic, Google Gemini, Groq, Azure, OpenRouter
- Zero config: Keys are pulled from the environment automatically
- SDK-first, not a framework: It stays out of your way.
β Not an AI framework
β Not an API gateway
β Not an agent memory system
It does one thing perfectly: Simplifying the API call to LLMs.
from ask_ai import OpenAI
ai = OpenAI()
# 1. Clean Markdown (Removes ```json and ``` tags)
# Perfect for extracting raw data from models that wrap everything in markdown
clean_text = ai.ask("Write JSON", clean=True).text
# 2. Extract Code (Returns ONLY the code block, ignores conversational filler)
# Great for automation pipelines
code = ai.ask("Write a python ping script", code=True).text
# 3. Strip Tags (Removes <think> blocks and HTML)
# Essential for reasoning models like DeepSeek-R1
answer_only = ai.ask("What is 1+1?", strip=True).text
# 4. Enforce & Parse JSON (Directly returns a Parsed Python Dictionary)
# Adds JSON instructions to the prompt and safely runs json.loads()
data_dict = ai.ask("Extract user info", json=True).json
print(data_dict['name'])
from ask_ai import OpenAI
ai = OpenAI()
# Will retry up to 3 times on RateLimit/Network errors, with generic timeout
response = ai.ask("Write a script", retry=3, timeout=10)Requests default to retry=2 (up to three attempts) for retryable timeout,
connection, rate-limit (429), and server (5xx) errors. Authentication and
invalid-request errors are not retried; use retry=0 to disable retries.
Streaming retries occur only before the first chunk/event, so already-emitted
output is never duplicated.
Set system prompts and temperature directly:
ai.advanced(
temperature=0.7,
prompt="You are a senior DevOps engineer."
)
print(ai.ask("How do I optimize a Dockerfile?").text)Unknown option names produce a UserWarning with a spelling suggestion. Pass
strict_options=True to a provider (or to an individual request) to raise a
ValueError instead. safe_mode currently has no defined behavior and is
deprecated; passing it emits a DeprecationWarning.
ask-ai works perfectly in asynchronous environments:
import asyncio
from ask_ai import OpenAI
async def main():
ai = OpenAI()
# Runs non-blocking in your event loop
response = await ai.ask_async("What is the capital of France?")
print(response.text)
asyncio.run(main())Stream responses token-by-token easily:
from ask_ai import OpenAI
ai = OpenAI()
# Sync Streaming
for chunk in ai.ask_stream("Write a short story about a blue dragon"):
print(chunk, end="", flush=True)
# Async Streaming
async def stream_example():
async for chunk in ai.ask_stream_async("Write a short story"):
print(chunk, end="", flush=True)Generate images or speech with compatible providers:
from ask_ai import OpenAI
ai = OpenAI()
# Generate an Image (DALL-E)
img_response = ai.ask("A majestic lion in a neon city", output_type="image")
img_response.save("lion.png")
# Generate Speech (TTS)
audio_response = ai.ask("Hello, this is a generated voice.", output_type="audio")
audio_response.save("welcome.mp3")Never experience downtime by supplying alternative fallback providers:
from ask_ai import OpenAI, Groq
ai = OpenAI()
# If OpenAI fails or rate-limits, it automatically switches to Groq
response = ai.ask("Explain quantum physics", providers=[ai, Groq])
print(response.text)Force LLMs to strictly respond matching a Pydantic schema:
from pydantic import BaseModel
from ask_ai import OpenAI
class User(BaseModel):
name: str
age: int
ai = OpenAI()
response = ai.ask("Extract name: Alice is 30 years old.", response_model=User)
user = response.pydantic
print(user.name) # "Alice"
print(user.age) # 30Structured responses support Pydantic models, dataclasses, TypedDict, and JSON
Schema dictionaries. The raw provider text remains available separately:
from dataclasses import dataclass
from ask_ai import OpenAI
@dataclass
class User:
name: str
age: int
response = OpenAI().ask(
"Extract: Alice is 30.",
response_model=User,
validation_retries=2,
)
print(response.raw)
print(response.parsed.name)For supported schemas, response_model also requests native JSON Schema output
from OpenAI/Azure, Anthropic, Groq, Google, and OpenRouter. When an adapter
receives a clear unsupported-schema response (HTTP 400/422), it retries once
without the native schema and keeps local validation enabled. Authentication,
rate-limit, and server errors are propagated, not retried as prompt-only
fallbacks. Schemas that cannot be represented in the provider's strict schema
format use prompt instructions and local validation directly. Provider/model
support varies; these request parameters have mock-based tests but have not
been verified against live APIs.
Provider parameters follow their documented APIs: OpenAI Chat Completions, Azure OpenAI, Anthropic, Groq, Google GenAI, and OpenRouter Chat Completions.
Use typed events for text, completion, and errors while keeping the legacy
ask_stream() API unchanged:
from ask_ai import Completed, OpenAI, TextDelta
for event in OpenAI().ask_stream_events("Tell me a story"):
if isinstance(event, TextDelta):
print(event.text, end="")
elif isinstance(event, Completed):
print("\nDone")UsageUpdate events are emitted when an adapter receives usage in its stream.
ToolCall is a reserved event type and is not emitted by the current adapters.
Streaming usage extraction is mock-tested and has not been verified against
live provider APIs.
Attach optional middleware to receive provider, model, latency, retry, and
provider-reported token usage. Usage field names are normalized to
input_tokens, output_tokens, and total_tokens. Estimated cost is only
calculated when usage is available and pricing for the model is supplied;
missing usage never produces a cost estimate. Mock tests cover usage extraction
for supported adapters; live provider APIs have not been verified. On failures,
middleware emits a failed event with failure_reason, then the exception is
propagated:
from ask_ai import Observability, OpenAI
ai = OpenAI().use(Observability(on_event=print))
response = ai.ask("Explain tracing")
print(response.metadata.latency_ms)
print(response.metadata.usage)Inspect advisory capability metadata before selecting a provider:
from ask_ai import OpenAI
capabilities = OpenAI().capabilities()
if capabilities.structured_output:
print("Structured output is available")Capabilities are advisory provider-level metadata, not guarantees for every model.
Route requests across providers using deterministic cheapest, fastest, or
most_reliable policies. Supply static profiles for meaningful ordering:
from ask_ai import Groq, OpenAI, Router
openai = OpenAI() # Requires OPENAI_API_KEY
groq = Groq() # Requires GROQ_API_KEY
router = Router(
[openai, groq],
policy="fastest",
profiles={
"OpenAIProvider": {"latency_ms": 100},
"GroqProvider": {"latency_ms": 50},
},
)
response = router.ask("Explain routing")Provider media downloads can be constrained by HTTPS, host, detected MIME, size,
and timeout policy. DNS results resolving to private, loopback, link-local, or
reserved IP addresses are rejected by default. allowed_hosts is optional;
set it for an explicit hostname allowlist. Only opt in to allow_private=True
when downloading from trusted private-network hosts. Redirects are rejected:
from ask_ai import MediaPolicy, download_media
data = download_media(
"https://example.com/image.png",
MediaPolicy(
max_bytes=10 * 1024 * 1024,
allowed_mime_types={"image/png"},
allowed_hosts={"example.com"},
),
)Create optional in-memory conversation state without changing the normal provider API:
session = OpenAI().session(max_messages=20)
session.ask("My name is Ada.")
print(session.ask("What is my name?").text)Session requests do not support response_model validation yet.
- Provider Fallback chaining & Pydantic Structured Output support
- Multi-Modal Vision input (image to text) for all key providers
- Native support for tools/function calling execution
- Optional in-memory session-based conversation management
[x] OpenAI, Groq, Gemini, Anthropic, Azure, OpenRouter support[x] Text to Image (DALL-E) and Text to Audio[x] Automated Retry & Timeout controls[x] Async Support (await ask_async)[x] Streaming Support (ask_stream/ask_stream_async)[x] Provider Fallback (ask(..., providers=[OpenAI, Groq]))[x] Structured Output (Pydantic Support)
- PyPI: askai-python
- GitHub Repository: Hosseinghorbani0/ask-ai (Star us! β)
- Official Website: hosseinghorbani0.ir
- Bug Tracker: Report an Issue