Lightweight, unified async client for OpenAI, Gemini, Mistral, Grok, Anthropic, and any OpenAI-compatible API. Single runtime dependency (niquests), automatic HTTP/2 and HTTP/3 negotiation.
For the full interactive CLI/TUI, use the separate padwan-cli package.
pip install padwan-llm
python
from padwan_llm import LLMClient
async with LLMClient(model="gpt-4o") as client:
response, usage = await client.complete_chat(
[{"role": "user", "content": "Hello!"}]
)
print(response["content"])
python
from padwan_llm import LLMClient, ConversationState
state = ConversationState(system="You are a concise assistant.")
async with LLMClient(model="gpt-4o") as client:
state.add_user_message("What's Python?")
stream = client.stream_chat(state.messages)
chunks: list[str] = []
async for text in stream:
print(text, end="", flush=True)
chunks.append(text)
state.add_assistant_message("".join(chunks))
if stream.usage:
state.accumulate_usage(stream.usage)
AgentSession drives a multi-turn conversation that can dispatch tool calls on each
round, feed the results back, and repeat until the model returns a plain text answer.
The mcp_tools list accepts both individual McpTool instances and whole
McpTransport servers — transports are entered as part of the session lifecycle:
from padwan_llm import AgentSession, LLMClient, McpStdio
async with AgentSession(
client=LLMClient(model="gpt-4o"),
mcp_tools=[McpStdio(command="uvx", args=["my-mcp-server"])],
system="You have access to tools. Use them when helpful.",
) as session:
async for chunk in session.stream("What's the weather in Paris?"):
print(chunk, end="", flush=True)
text = await session.send("And in London?")
AgentSession supports sequential or parallel tool execution, approval hooks,
per-tool error handlers, and optional snapshot persistence via a
ConversationStore protocol — see docs/agents.md.
Both streamable-HTTP and stdio MCP transports are built in:
from padwan_llm import McpStreamable, McpStdio
async with McpStreamable(url="https://mcp.example.com/mcp", token="sk-...") as mcp:
for tool in mcp.tools:
print(tool.name, tool.description)
async with McpStdio(command="uvx", args=["my-mcp-server"]) as mcp:
result = await mcp.tools[0].handler({"query": "hello"})
See docs/mcp.md for the full feature matrix and architecture.
Gemini's reasoning models can stream their internal thought tokens separately from
the final answer. Wire an on_thought callback to receive them:
from padwan_llm import GeminiClient
thoughts: list[str] = []
async with GeminiClient(
model="gemini-2.5-flash",
on_thought=thoughts.append,
thinking_config={"thinkingBudget": 2048, "includeThoughts": True},
) as client:
stream = client.stream_chat([{"role": "user", "content": "What is 7 * 8?"}])
async for chunk in stream:
print(chunk, end="")
print("\n---\nReasoning:", "".join(thoughts))
RealtimeClient opens a bidirectional voice session over a WebSocket and yields
the live connection: stream microphone audio in, receive model audio and
transcripts back. OpenAI (gpt-realtime), Gemini Live, and Grok Voice are
supported, dispatched by model name. Requires the realtime extra
(pip install "padwan-llm[realtime]"):
from padwan_llm import RealtimeClient
async with RealtimeClient(instructions="Answer briefly.", voice="marin") as conn:
await conn.append_audio(pcm16_chunk) # mono PCM16 microphone audio
async for event in conn:
if audio := conn.audio_delta_bytes(event):
playback.write(audio)
Server-side VAD drives turn-taking by default; pass turn_detection=NO_TURN_DETECTION
for manual push-to-talk. See the realtime sections of
docs/clients/openai.md,
docs/clients/gemini.md, and
docs/clients/grok.md.
Opt-in GenAI spans and metrics for every provider client, following the OTel
GenAI semantic conventions. Requires the otel extra
(pip install "padwan-llm[otel]"):
from padwan_llm import otel
otel.instrument() # uses the global tracer/meter providers
For a managed trace backend, the Langfuse adapter configures both sides and maps Padwan chat, agent, tool, embedding, and MCP spans to Langfuse observations:
pip install "padwan-llm[langfuse]"
python
from padwan_llm.langfuse import instrument
telemetry = instrument() # uses the standard LANGFUSE_* environment variables
Chat calls emit a chat <model> client span (provider, model, server address,
token usage including reasoning tokens, thinking duration, finish reasons,
requested tool names) plus the gen_ai.client.operation.duration and
gen_ai.client.token.usage histograms. Agent tool execution emits
execute_tool spans; embeddings, batch operations, and realtime sessions get
their own spans. See docs/observability.md for the
full attribute list.
just e2e-otel runs the e2e suite against a local Grafana stack with a
ready-made GenAI dashboard
(bin/observability/dashboards):
export OPENAI_API_KEY=...
padwan-llm "Hello!" -m gpt-4o-mini
uvx padwan-llm "Hello!" -m gpt-4o-mini
Auto-detected providers: OpenAI, Gemini, Mistral, Grok, Anthropic (claude-*).
Any OpenAI-compatible API (Groq, Together AI, Ollama, vLLM, ...) is supported via OpenAIClient with a custom base_url.
Unit tests run by default (no API keys needed):
uv run pytest
E2e tests require API keys. Create a .env file or pass one with --env-file:
uv run pytest tests/e2e/ -m e2e
uv run pytest tests/e2e/ -m e2e --env-file path/to/.env
Tests for providers whose API key is missing are automatically skipped.
OPENAI_API_KEY=...
GEMINI_API_KEY=...
MISTRAL_API_KEY=...
GROK_API_KEY=...
ANTHROPIC_API_KEY=...
Aggregators that expose OSS variants of many model families behind a single OpenAI-compatible endpoint and token are supported with two env vars — every model then routes through that gateway, with no per-provider keys or per-call overrides:
PADWAN_BASE_URL=https://your-gateway.example.com/v1/
PADWAN_API_KEY=...
async with LLMClient(model="gemini-2.5-flash") as client:
response, usage = await client.complete_chat([{"role": "user", "content": "Hi!"}])
Precedence is explicit base_url/ api_key args → PADWAN_* → native
per-provider env vars. Passing an explicit base_url disables gateway mode and
restores native provider routing.