Requisite

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?")

Install

Requisite is published on PyPI:

pip install requisite-ai                  # core only -- no provider SDKs
pip install "requisite-ai[all]"           # every provider + langgraph
pip install "requisite-ai[openai]"        # OpenAI only
pip install "requisite-ai[anthropic]"     # Anthropic (Claude) only
pip install "requisite-ai[gemini]"        # Gemini only
pip install "requisite-ai[groq]"          # Groq only (uses the openai package -- wire-compatible)
pip install "requisite-ai[azure_openai]"  # Azure OpenAI only (uses the openai package)
pip install "requisite-ai[mcp]"           # MCP client integration
pip install "requisite-ai[rag]"           # RAG (embedding providers; in-memory vector store needs nothing extra)
pip install "requisite-ai[langgraph]"     # native + langgraph orchestration

Quoting the package name ("requisite-ai[all]") avoids shell globbing issues with [...] in zsh; plain pip install requisite-ai[all] also works in bash.

Contributing to Requisite itself (not just using it)? See CONTRIBUTING.md for an editable install from source (pip install -e ".[dev,all]") and running the test suite/linters.

Note on the Gemini SDK: this framework uses the current, unified google-genai package (from google import genai). Do not install the deprecated google-generativeai package — the two conflict.

Note on Azure OpenAI: uses Azure's current v1 GA API (the plain openai client pointed at your endpoint) — no separate SDK, no dated api-version string. See ADR-0002.

Configuration

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.2
RATE_LIMIT_RPM=          # optional -- see "Rate limiting" below

Settings (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.

Usage

Chat

from requisite import AI

ai = AI()
print(ai.chat("Explain LangGraph in one sentence."))

Supported providers

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.

Structured output

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 30

Tool calling

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}"

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).

Agents

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.

Multi-agent workflows

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 engine

Let a supervisor agent delegate to a team of workers, addressed by name, deciding when the task is done:

supervisor = Agent(name="Supervisor", provider="openai")
researcher = Agent(name="Researcher", provider="openai")
writer = Agent(name="Writer", provider="openai")

workflow = Workflow().supervisor()
workflow.add(supervisor).add(researcher).add(writer)
result = workflow.run("Research AI trends and write a summary.")

The first agent added is the coordinator; every agent added after it is a worker. .planner() works the same way, but the first agent decomposes the task into an ordered plan up front instead of delegating round by round. .reflection() takes a single agent that critiques and revises its own output over several rounds:

workflow = Workflow().reflection()
workflow.add(writer)
result = workflow.run("Write a tagline for an AI framework.", max_rounds=3)

Rate limiting

Free-tier and other quota-limited API keys often cap requests per minute — set RATE_LIMIT_RPM (and, optionally, RATE_LIMIT_MAX_WAIT_SECONDS) in .env and AI/Agent wait for capacity instead of letting the provider reject the call:

RATE_LIMIT_RPM=15

That covers a single Agent/AI. Several agents that call the same underlying API key share the same real quota, so build one RateLimiter and pass it to each of them explicitly — this is what a multi-agent Workflow (research + writer + planner + supervisor, say) needs to avoid exceeding the quota collectively even if each individual agent looks fine on its own:

from requisite import Agent, RateLimiter

shared_limit = RateLimiter(requests_per_minute=15)
research = Agent(name="Researcher", provider="gemini", rate_limiter=shared_limit)
writer = Agent(name="Writer", provider="gemini", rate_limiter=shared_limit)

Skills

A 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()])

Capabilities: declare what, not which

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.

MCP integration

Connect to any MCP server (local via stdio, or remote via Streamable HTTP) and its tools become capabilities like any other:

from requisite import Agent
from requisite.mcp import MCPClient
from requisite.capabilities import default_registry as capabilities

# Local, subprocess-based MCP server
filesystem = MCPClient.stdio(
    name="filesystem",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "/path/to/allow"],
)
filesystem.register_as_capability(capabilities, capability="read_file")

# Remote MCP server
github = MCPClient.http(name="github", url="https://api.example.com/mcp", headers={"Authorization": "Bearer ..."})
github.register_as_capability(capabilities, capability="github")

agent = Agent(name="Assistant", provider="openai")
agent.requires("read_file", "github")  # agent can't tell these apart from native tools

You can also use an MCP server's tools directly, without going through capabilities:

tools = filesystem.discover_tools()
agent = Agent(name="Assistant", provider="openai", tools=tools)

Both transports were verified against real MCP servers during development; each tool call re-connects, calls, and disconnects rather than holding a persistent session — see ADR-0004 for why, and the plan for an optional persistent-session mode if reconnect latency becomes a problem for your use case.

RAG (Retrieval-Augmented Generation)

A retriever is a capability too — agent.requires("knowledge_base") works exactly like "weather" or an MCP-backed tool:

from requisite import Agent
from requisite.rag import Retriever
from requisite.rag.embeddings import OpenAIEmbeddingProvider
from requisite.rag.vectorstores import InMemoryVectorStore
from requisite.capabilities import default_registry as capabilities

retriever = Retriever(
    embedding_provider=OpenAIEmbeddingProvider(api_key="sk-..."),
    vector_store=InMemoryVectorStore(),  # zero-dependency default
)
retriever.add_texts([
    "Paris is the capital of France.",
    "The Eiffel Tower was completed in 1889.",
])

capabilities.register("knowledge_base", retriever.as_tool())

agent = Agent(name="Assistant", provider="openai")
agent.requires("knowledge_base")
print(agent.run("What's the capital of France?").content)

add_texts chunks each document (character-based, with overlap) before embedding and storing it:

retriever.add_texts(long_document_text, chunk_size=1000, chunk_overlap=200)

InMemoryVectorStore is the zero-dependency default — pure-Python cosine similarity, fine for a few thousand chunks. Pinecone and Weaviate integrations are planned (PINECONE_API_KEY/WEAVIATE_URL are already reserved in .env.example) but not yet implemented — see ADR-0005 for the interface decomposition (embeddings / vector stores / retrievers are three independent extension points) and why the in-memory default was chosen over requiring a real vector DB from day one.

Memory: conversation history across separate calls

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.

Conversation policies: keeping long histories bounded

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.

Prompt templates

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.

Structured logging

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.

Streaming & async

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.")

Conversation history

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.

Architecture

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
├── mcp/            # BaseMCPClient + MCPClient (stdio + Streamable HTTP)
│                   # + MCPClientRegistry -- bridges MCP tools into capabilities
├── rag/            # BaseEmbeddingProvider, BaseVectorStore, BaseRetriever
│                   # + Retriever (dense) + InMemoryVectorStore
├── 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,
│                   # reflection, planner, supervisor) 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.

Error handling

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              # an MCP server call/discovery failed, or a capability bridge found no matching tool

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.

Testing

pytest

Tests 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.

Roadmap

Implemented: 5 providers (OpenAI, Anthropic, Gemini, Groq, Azure OpenAI), structured outputs, tool calling, skills, capability resolution (agent.requires(...)), MCP client integration (stdio + Streamable HTTP), RAG (embeddings, in-memory vector store, dense retrieval — Pinecone/Weaviate still planned), memory + conversation policies (Agent(memory=..., conversation_policy=...)), prompt templates, structured logging, agents + registry, multi-agent workflows (sequential, parallel, reflection, planner, supervisor on the native backend; sequential on langgraph).

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.

Contributing

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 the requisite-core vs. 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.

License

MIT — see LICENSE.