Building Custom Providers

When the built-in providers don't fit, subclass ContextProvider. The base class handles tool wrapping, name derivation, and error shaping.

Minimal Example

Create a virtual environment using SDK setup, then install the model dependency and set your key:

uv pip install -U agno openai
export OPENAI_API_KEY="your-openai-api-key"

Save as faq_provider.py and run python faq_provider.py. The FAQ data is a local illustration.

import asyncio

from agno.agent import Agent
from agno.models.openai import OpenAIResponses
from agno.context import Answer, ContextProvider, Status

FAQ = {"pricing": "See agno.com/pricing", "support": "Email support@agno.com"}

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, *, run_context=None) -> 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, *, run_context=None) -> Answer:
        return self.query(question, run_context=run_context)

faq = FAQContextProvider(id="faq")
agent = Agent(model=OpenAIResponses(id="gpt-5.4-mini"), tools=faq.get_tools())

if __name__ == "__main__":
    asyncio.run(agent.aprint_response("How do I contact support?"))

The agent now has a query_faq tool. Same shape as every built-in provider.

Required Methods

You must implement these four abstract methods:

Method	Purpose
`query(question, *, run_context=None) -> Answer`	Sync read
`aquery(question, *, run_context=None) -> Answer`	Async read
`status() -> Status`	Sync health check
`astatus() -> Status`	Async health check

Answer

Answer is what query() returns. These fragments belong inside a query method:

from agno.context import Answer, Document

# Text-only answer
return Answer(text="The weather is sunny.")

# Answer with source documents
return Answer(
    text="Found 3 matching policies.",
    results=[
        Document(id="doc1", name="Refund Policy", uri="/policies/refund.md", snippet="..."),
        Document(id="doc2", name="Privacy Policy", uri="/policies/privacy.md", snippet="..."),
    ]
)

Status

Status reports provider health. These fragments belong inside a status method:

from agno.context import Status

# Healthy
return Status(ok=True, detail="Connected to database")

# Unhealthy
return Status(ok=False, detail="API key invalid")

Optional Methods

Override these to customize behavior:

Method	Default	Override when
`update()` / `aupdate()`	Raises `NotImplementedError`	Provider supports writes
`asetup()`	No-op	Need async init (MCP sessions, cache priming)
`aclose()`	No-op	Hold long-lived state (watches, connections)
`instructions()`	Generic guidance	Want source-specific usage hints

Adding Write Support

Override update(), aupdate(), and _default_tools() for writable providers:

class NotesContextProvider(ContextProvider):
    def __init__(self, id: str, storage: dict):
        super().__init__(id)
        self.storage = storage

    def query(self, question: str, *, run_context=None) -> Answer:
        matches = [v for k, v in self.storage.items() if question.lower() in k.lower()]
        return Answer(text="\n".join(matches) if matches else "No matching notes.")

    async def aquery(self, question: str, *, run_context=None) -> Answer:
        return self.query(question, run_context=run_context)

    def update(self, instruction: str, *, run_context=None) -> Answer:
        if instruction.startswith("save note:"):
            parts = instruction[10:].split(" - ", 1)
            if len(parts) == 2:
                self.storage[parts[0].strip()] = parts[1].strip()
                return Answer(text=f"Saved note: {parts[0].strip()}")
        return Answer(text="Could not parse instruction. Use: save note: <title> - <content>")

    async def aupdate(self, instruction: str, *, run_context=None) -> Answer:
        return self.update(instruction, run_context=run_context)

    def _default_tools(self) -> list:
        return self._read_write_tools()  # Exposes both query and update tools

    def status(self) -> Status:
        return Status(ok=True, detail=f"{len(self.storage)} notes")

    async def astatus(self) -> Status:
        return self.status()

Continue the FAQ script with this class and replace its agent construction with:

notes = NotesContextProvider(id="notes", storage={})
agent = Agent(model=OpenAIResponses(id="gpt-5.4-mini"), tools=notes.get_tools())

The agent now has query_notes and update_notes. This dictionary is a process-local demonstration; choose durable storage for a deployed service.

This partial class illustrates setup and teardown; implement the remaining abstract methods before instantiating it. Install aiohttp for this example. Call asetup() before queries and aclose() in finally:

class StreamingAPIContextProvider(ContextProvider):
    def __init__(self, id: str, api_url: str):
        super().__init__(id)
        self.api_url = api_url
        self.session = None

    async def asetup(self) -> None:
        import aiohttp
        self.session = aiohttp.ClientSession()

    async def aclose(self) -> None:
        if self.session:
            await self.session.close()

    async def aquery(self, question: str, *, run_context=None) -> Answer:
        async with self.session.get(f"{self.api_url}/search", params={"q": question}) as resp:
            data = await resp.json()
            return Answer(text=data.get("answer", "No answer found."))

    # ... implement query, status, astatus

Custom Instructions

Override instructions() to provide source-specific guidance:

def instructions(self) -> str:
    return """
    Use query_jira for:
    - Finding issues by key (e.g., "PROJ-123")
    - Searching by assignee, status, or labels
    - Getting sprint information

    Use update_jira for:
    - Changing issue status
    - Adding comments
    - Updating assignee
    """

Using RunContext

The run_context parameter carries caller state. This method fragment assumes your application has authenticated the caller and implemented get_docs_for_user and search. Choose an explicit missing-user policy; the illustrative fallback below exposes only public documents:

def query(self, question: str, *, run_context=None) -> Answer:
    user_id = run_context.user_id if run_context else None

    if user_id:
        # Fetch user-specific data
        user_docs = self.get_docs_for_user(user_id)
        return Answer(text=self.search(question, user_docs))

    # Fall back to global search
    return Answer(text=self.search(question, self.public_docs))

Available on run_context:

Field	Description
`user_id`	Identifies the caller
`session_id`	Identifies the conversation
`metadata`	Arbitrary dict passed through the call chain
`dependencies`	Values injected via the agent's `dependencies` parameter

Wrapping External APIs

This illustrative contract uses a fictional endpoint, not a real weather API. Replace the base URL, authentication, request fields and response parsing with your service’s documented contract. Install httpx. Both HTTP clients are scoped to their calls, and async methods perform async I/O:

import httpx
from agno.context import Answer, ContextProvider, Status

class WeatherContextProvider(ContextProvider):
    def __init__(self, id: str, api_key: str, base_url: str = "https://example.invalid"):
        super().__init__(id, write=False)
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")

    def _params(self, question: str) -> dict:
        return {"city": question.removeprefix("weather in").strip(), "key": self.api_key}

    def _answer(self, response: httpx.Response) -> Answer:
        response.raise_for_status()
        data = response.json()
        return Answer(text=f"{data['temp']}F, {data['condition']}")

    def query(self, question: str, *, run_context=None) -> Answer:
        with httpx.Client(timeout=10) as client:
            return self._answer(client.get(f"{self.base_url}/current", params=self._params(question)))

    async def aquery(self, question: str, *, run_context=None) -> Answer:
        async with httpx.AsyncClient(timeout=10) as client:
            return self._answer(await client.get(f"{self.base_url}/current", params=self._params(question)))

    def status(self) -> Status:
        try:
            with httpx.Client(timeout=10) as client:
                client.get(f"{self.base_url}/health").raise_for_status()
            return Status(ok=True, detail="API reachable")
        except httpx.HTTPError as exc:
            return Status(ok=False, detail=str(exc))

    async def astatus(self) -> Status:
        try:
            async with httpx.AsyncClient(timeout=10) as client:
                (await client.get(f"{self.base_url}/health")).raise_for_status()
            return Status(ok=True, detail="API reachable")
        except httpx.HTTPError as exc:
            return Status(ok=False, detail=str(exc))

Provider Catalog

Browse all built-in providers for inspiration