Custom Context Provider
When a built-in provider doesn't fit, subclass `ContextProvider`.
The source omits the keyword-only run_context parameter accepted by ContextProvider.query() and ContextProvider.aquery(). The generated query_faq tool calls aquery(..., run_context=...), so it returns a serialized TypeError instead of the FAQ answer. Update both signatures before running.
"""
Custom Context Provider
=======================
When a built-in provider doesn't fit, subclass `ContextProvider`. The
ABC handles tool wrapping, name derivation, and error shaping — you
only write `aquery` + `astatus`.
Here: a tiny FAQ source over an in-memory dict. The agent calls
`query_faq(question)` and gets the matching answer back.
Requires: OPENAI_API_KEY
"""
from __future__ import annotations
import asyncio
from agno.agent import Agent
from agno.context import Answer, ContextProvider, Status
from agno.models.openai import OpenAIResponses
# ---------------------------------------------------------------------------
# The data
# ---------------------------------------------------------------------------
FAQ = {
"return": "Returns accepted within 30 days. Email support@example.com.",
"hours": "We're open Mon-Fri, 9am-5pm ET.",
"shipping": "Orders ship in 2-3 business days via USPS.",
}
# ---------------------------------------------------------------------------
# The provider
# ---------------------------------------------------------------------------
class FAQContextProvider(ContextProvider):
def status(self) -> Status:
return Status(ok=True, detail=f"{len(FAQ)} entries")
async def astatus(self) -> Status:
return self.status()
def query(self, question: str) -> Answer:
key = next((k for k in FAQ if k in question.lower()), None)
return Answer(text=FAQ[key] if key else "No FAQ entry matches that.")
async def aquery(self, question: str) -> Answer:
return self.query(question)
# ---------------------------------------------------------------------------
# Wire it into an agent
# ---------------------------------------------------------------------------
faq = FAQContextProvider(id="faq")
agent = Agent(
model=OpenAIResponses(id="gpt-5.4"),
tools=faq.get_tools(),
instructions=faq.instructions(),
markdown=True,
)
if __name__ == "__main__":
asyncio.run(agent.aprint_response("What's your return policy?"))Run the Example
Set up your virtual environment
uv venv --python 3.12
source .venv/bin/activateInstall dependencies
uv pip install -U agno openaiExport your OpenAI API key
export OPENAI_API_KEY="your_openai_api_key_here"Accept the run context
Add from agno.run import RunContext. Change query to def query(self, question: str, *, run_context: RunContext | None = None) -> Answer: and aquery to async def aquery(self, question: str, *, run_context: RunContext | None = None) -> Answer:. In aquery, return self.query(question, run_context=run_context).
Run the example
Save the code above as custom_provider.py, then run:
python custom_provider.pyFull source: cookbook/12_context/10_custom_provider.py