requisite-ai 0.1.0

A provider-agnostic, plugin-based framework for building AI applications and agents. Declare what you need -- providers, tools, capabilities -- not which SDK provides it.

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

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 orchestration

Or with the plain requirements file: pip install -r requirements.txt

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

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

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.

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

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.

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
├── 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.

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

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

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.