Declare what your AI application needs — not which SDK provides it.
A provider-agnostic, plugin-based Python framework for building AI
applications and agents. Swap the LLM provider, the multi-agent execution
engine, or the implementation behind a capability like "weather" or
"internet_search" — all via configuration, never a rewrite.
from requisite import AI
ai = AI() # provider="openai" by default
ai = AI(provider="anthropic", model="claude-sonnet-4-6") # same API, different provider
ai = AI(provider="gemini", model="gemini-2.5-flash")
ai = AI(provider="groq", model="llama-3.3-70b-versatile")from requisite import Agent
agent = Agent(name="Assistant", provider="openai")
agent.requires("weather", "internet_search", "filesystem") # not use_tool(specific_impl)
agent.run("What's the weather in Tokyo?")pip install -e .[all] # every provider + langgraph
pip install -e .[openai] # OpenAI only
pip install -e .[anthropic] # Anthropic (Claude) only
pip install -e .[gemini] # Gemini only
pip install -e .[groq] # Groq only (uses the openai package -- wire-compatible)
pip install -e .[azure_openai] # Azure OpenAI only (uses the openai package)
pip install -e .[langgraph] # native + langgraph orchestrationOr with the plain requirements file: pip install -r requirements.txt
Note on the Gemini SDK: this framework uses the current, unified
google-genaipackage (from google import genai). Do not install the deprecatedgoogle-generativeaipackage — the two conflict.Note on Azure OpenAI: uses Azure's current v1 GA API (the plain
openaiclient pointed at your endpoint) — no separate SDK, no datedapi-versionstring. See ADR-0002.
Copy .env.example to .env and fill in the key(s) for the provider(s) you use:
OPENAI_API_KEY=
ANTHROPIC_API_KEY=
GEMINI_API_KEY=
GROQ_API_KEY=
AZURE_OPENAI_API_KEY=
AZURE_OPENAI_ENDPOINT=
DEFAULT_PROVIDER=openai
MODEL=gpt-4o-mini
TEMPERATURE=0.2Settings (requisite/config/settings.py) reads this automatically — you
never call os.environ.get yourself. .env.example also reserves
placeholders for planned integrations (GitHub, Hugging Face, AWS, Azure
general-purpose credentials, Pinecone, Weaviate) — see ROADMAP.md.
from requisite import AI
ai = AI()
print(ai.chat("Explain LangGraph in one sentence."))| Provider | provider= |
Model examples | Notes |
|---|---|---|---|
| OpenAI | "openai" |
gpt-4o-mini, gpt-4o |
|
| Anthropic | "anthropic" |
claude-sonnet-4-6, claude-opus-4-8 |
Native structured output via messages.parse |
| Gemini | "gemini" |
gemini-2.5-flash, gemini-2.5-pro |
Uses the unified google-genai SDK |
| Groq | "groq" |
llama-3.3-70b-versatile, openai/gpt-oss-20b |
OpenAI-wire-compatible; uses the openai package |
| Azure OpenAI | "azure_openai" |
your deployment name | Requires azure_endpoint (or AZURE_OPENAI_ENDPOINT); current v1 GA API, no api-version needed |
Switching between any of these is the provider=/AZURE_OPENAI_ENDPOINT
change shown above — no other code changes. See
ADR-0002 for
why Groq and Azure OpenAI are implemented as thin OpenAIProvider
subclasses rather than separate SDKs.
from pydantic import BaseModel
class Person(BaseModel):
name: str
age: int
person = ai.chat("Extract: John is 30 years old.", response_model=Person)
print(person.name, person.age) # John 30from requisite.tools import tool
@tool
def get_weather(city: str) -> str:
"""Get the current weather for a city."""
return f"Sunny, 22C in {city}"
response = ai.chat_response("What's the weather in Paris?", tools=[get_weather])
if response.has_tool_calls:
call = response.tool_calls[0]
result = get_weather.tool.execute(**call.arguments)That's the low-level view. For most applications, let an Agent run the
full tool-calling loop for you (see below).
from requisite import Agent
from requisite.tools import tool
@tool
def get_weather(city: str) -> str:
"""Get the current weather for a city."""
return f"Sunny, 22C in {city}"
agent = Agent(name="Weather Agent", provider="openai", tools=[get_weather])
result = agent.run("What's the weather in Paris?")
print(result.content) # "It's sunny and 22C in Paris."
print(result.tool_calls_executed) # ["get_weather"]Agent automatically: offers its tools/skills to the model, executes any
tool calls the model requests, feeds results back, and repeats (up to
max_iterations) until it has a final answer.
from requisite import Agent, Workflow
research = Agent(name="Researcher", provider="openai")
writer = Agent(name="Writer", provider="openai")
workflow = Workflow()
workflow.add(research)
workflow.add(writer)
result = workflow.run("Research AI trends and write a summary.")
print(result.content)Run agents in parallel against the same input instead of as a pipeline:
workflow.parallel()
result = workflow.run("What is retrieval-augmented generation?")Switch the execution engine — the .add() / .run() API never changes:
workflow.use_langgraph() # requires: pip install langgraph
result = workflow.run("Research AI trends and write a summary.")
workflow.use_native() # back to the built-in, dependency-free engineA skill is a reusable, higher-level capability (vs. a tool, which is typically a single function). Skills expose themselves to the model as tools automatically:
from requisite.skills import BaseSkill
class ReadFileSkill(BaseSkill):
def __init__(self):
super().__init__(name="read_file", description="Read a text file's contents.")
def run(self, path: str) -> str:
with open(path) as f:
return f.read()
agent = Agent(name="File Agent", provider="openai", skills=[ReadFileSkill()])agent.use_tool(specific_impl) binds an agent to one concrete
implementation at write-time. agent.requires("weather") binds it to a
name, resolved at runtime against whichever implementation is
currently available — a native tool, an MCP server, a cloud API, or a
third-party plugin:
from requisite import Agent
agent = Agent(name="Assistant", provider="openai")
agent.requires("weather", "internet_search", "filesystem")
result = agent.run("What's the weather in Tokyo?")Three capabilities ship out of the box, backed by free/keyless APIs and
the local filesystem (see requisite/capabilities/resolvers.py):
"filesystem", "weather", "internet_search".
Register a better provider for the same capability at a higher priority and it takes over automatically — application code never changes:
from requisite.capabilities import default_registry
default_registry.register(
"weather",
my_paid_weather_tool,
provider_name="acme-weather",
priority=10,
is_available=lambda: bool(os.environ.get("ACME_API_KEY")),
)
# agent.requires("weather") now resolves to "acme-weather" when the key
# is set, and quietly falls back to the built-in provider otherwise.This is the same interface + registry pattern used for providers and
orchestrators, one layer up: CapabilityRegistry.resolve(name) picks the
highest-priority provider whose is_available() currently returns
True, raising CapabilityException if none are.
from requisite import Agent
from requisite.memory import InProcessMemory
memory = InProcessMemory()
agent = Agent(name="Assistant", provider="openai", memory=memory, session_id="user-42")
agent.run("My name is Alex.")
result = agent.run("What's my name?") # remembers "Alex" via `memory`
print(result.content)session_id is required whenever memory is set — there's no implicit
"current user," so an agent with memory but no session_id fails at
construction time rather than silently sharing one conversation across
callers. When memory is configured, run()/arun() must be called with
a plain string (the new turn); prior history is loaded from memory
automatically. Only the user's turn and the agent's final answer are
persisted, not intermediate tool-call round-trips — see
ADR-0002 for
the reasoning.
InProcessMemory (dict-backed, lost on restart) is the zero-dependency
default. Implement requisite.memory.base.BaseMemory for a persistent
backend (Redis, SQLite, ...) — see ROADMAP.md.
A conversation that grows unbounded eventually blows past the model's
context window (or just gets expensive). conversation_policy= trims or
summarizes history once, before each run()/arun() call — independent
of whether memory is configured:
from requisite import Agent
from requisite.memory import InProcessMemory, MessageCountPolicy
agent = Agent(
name="Assistant",
provider="openai",
memory=InProcessMemory(),
session_id="user-42",
conversation_policy=MessageCountPolicy(max_messages=20), # keep the most recent 20
)For long-running conversations where you'd rather compress old context
than drop it, SummarizingPolicy collapses older messages into one
LLM-generated summary, keeping the most recent few verbatim:
from requisite import AI
from requisite.memory import SummarizingPolicy
# A separate, cheaper AI instance for summarization is a reasonable choice --
# summarization quality requirements are usually lower than the agent's own task.
summarizer = AI(provider="groq", model="llama-3.3-70b-versatile")
agent = Agent(
name="Assistant",
provider="openai",
memory=InProcessMemory(),
session_id="user-42",
conversation_policy=SummarizingPolicy(summarizer, max_messages=20, keep_recent=6),
)See ADR-0003
for why the policy is applied once per call rather than mid-tool-loop,
and why it doesn't change what gets persisted to memory.
from requisite.prompts import ChatPromptTemplate
chat_template = ChatPromptTemplate.from_messages([
("system", "You are a {persona}."),
("user", "{question}"),
])
messages = chat_template.format_messages(persona="pirate", question="Where's the treasure?")
print(ai.chat(messages))ChatPromptTemplate.format_messages(...) returns a plain list[Message]
— pass it to ai.chat(...) or agent.run(...) exactly like any other
message sequence. For a single string instead of a full conversation,
use PromptTemplate:
from requisite.prompts import PromptTemplate
translate = PromptTemplate.from_template("Translate to {language}: {text}")
print(ai.chat(translate.format(language="French", text="Good morning")))
# Pre-fill some variables now, leave the rest for later:
french_translator = translate.partial(language="French")
print(ai.chat(french_translator.format(text="Good night")))Register named templates for reuse across an application with
PromptTemplateRegistry.
Every framework module logs through the standard library
(logging.getLogger("requisite.<subpackage>")). Opt into JSON output
with one call, wherever your application configures logging — this is
never done automatically by the framework:
from requisite.telemetry import configure_logging
configure_logging(level="INFO", json_format=True){"timestamp": "...", "level": "DEBUG", "logger": "requisite.capabilities", "message": "Resolved capability 'weather' -> 'open-meteo'", "capability": "weather", "provider_name": "open-meteo"}Any extra={...} fields passed to a log call are merged into the JSON
payload automatically — the same log call produces a readable line with
the default formatter and a structured payload with json_format=True.
for token in ai.stream("Write a haiku about distributed systems."):
print(token, end="")
text = await ai.achat("Hello!")
async for token in ai.astream("Hello, streamed!"):
print(token, end="")
result = await agent.arun("What's the weather in Paris?")
result = await workflow.arun("Research AI trends.")from requisite import Message
history = [
Message.user("My name is Alex."),
Message.assistant("Nice to meet you, Alex!"),
Message.user("What's my name?"),
]
print(ai.chat(history))See examples/ for complete, runnable scripts covering each of the above.
requisite/
├── core/ # Provider-agnostic data models (Message, ChatResponse,
│ # ToolCall, ...) and the AIException hierarchy
├── config/ # Settings (pydantic-settings, reads .env)
├── providers/ # BaseProvider interface + OpenAI, Anthropic, Gemini, Groq,
│ # Azure OpenAI + ProviderRegistry (extensible, DI-friendly)
├── tools/ # Tool, @tool decorator, ToolRegistry, JSON Schema derivation
├── skills/ # BaseSkill, SkillRegistry -- reusable higher-level capabilities
├── capabilities/ # CapabilityRegistry -- resolve a named capability (e.g.
│ # "weather") to whichever implementation is available
├── memory/ # BaseMemory + InProcessMemory + MemoryRegistry, plus
│ # BaseConversationPolicy (MessageCountPolicy, SummarizingPolicy)
├── prompts/ # PromptTemplate, ChatPromptTemplate, PromptTemplateRegistry
├── telemetry/ # Structured (JSON) logging -- opt-in, never automatic
├── agents/ # Agent (tool-calling loop, .requires(), optional memory) + AgentRegistry
├── orchestrators/ # BaseOrchestrator interface + native (sequential/parallel)
│ # and langgraph backends + OrchestratorRegistry
├── workflows/ # Workflow -- the small, ergonomic multi-agent facade
└── ai.py # The `AI` facade -- the class most users touch directly
Every layer follows the same pattern: a small abstract interface
(BaseProvider, BaseOrchestrator, ...), one or more concrete
implementations, and a plain, instantiable registry (not a singleton)
mapping names to constructors. That's what makes each of the following a
configuration change rather than a code change:
AI(provider="openai")→AI(provider="gemini")Workflow(orchestrator="native")→workflow.use_langgraph()agent.use_tool(specific_impl)→agent.requires("weather")
See ARCHITECTURE.md for the full dependency
diagram, request-flow walkthroughs (a chat call, an agent's tool-calling
loop, capability resolution, a multi-agent workflow), and the design
decisions behind them. See CONTRIBUTING.md for
the step-by-step to add a new provider, orchestrator backend, or
capability resolver.
All framework exceptions inherit from AIException:
AIException
├── ConfigurationException # missing/invalid config, unknown provider/orchestrator name
├── ProviderException # provider SDK call failed (wraps the original error)
├── ToolException # a tool wasn't found, or raised while executing
├── SkillException # a skill wasn't found, or raised while executing
├── AgentException # agent execution failed (e.g. max_iterations exceeded)
├── CapabilityException # a required capability has no available provider
├── PromptException # a prompt template was rendered without a required variable
└── MCPException # reserved for the upcoming MCP integration
Provider SDK errors are never swallowed — they're wrapped with provider
and original_error attached, and re-raised via raise ... from original_error
so the original traceback is preserved.
pytestTests never hit the network: the OpenAI and Gemini SDKs are faked via
sys.modules injection, and the AI / Agent / Workflow facades are
tested against fully in-memory fake providers.
Implemented: 5 providers (OpenAI, Anthropic, Gemini, Groq, Azure OpenAI),
structured outputs, tool calling, skills, capability resolution
(agent.requires(...)), memory + conversation policies
(Agent(memory=..., conversation_policy=...)), prompt templates,
structured logging, agents + registry, multi-agent workflows
(sequential/parallel, native + langgraph backends).
See ROADMAP.md for the full, per-layer status table
(providers, orchestration strategies, MCP, memory, RAG, ...) and what's
explicitly out of scope. See FEATURES.md for the same
information organized as a line-by-line checklist against the original
project vision.
Contributions are welcome:
CONTRIBUTING.md— dev setup, running checks, and step-by-step guides for adding a provider, orchestrator backend, or capability resolver.ARCHITECTURE.md— how the framework fits together and why: the interface + registry pattern, request-flow walkthroughs, design decisions.docs/adr/— Architecture Decision Records: the formal rationale behind core interfaces, extension points, plugin discovery, configuration model, and therequisite-corevs. optional-integrations boundary. Start with ADR-0001.DEVELOPMENT.md— coding standards, testing philosophy, docstring format, logging/error-handling conventions, versioning policy.ROADMAP.md— what's shipped, planned, or out of scope.
Please also read the Code of Conduct. Security
issues should go through SECURITY.md, not a public issue.
See CHANGELOG.md for release history.
MIT — see LICENSE.