KnowledgeProtocol

​

Overview

• What tools it exposes to the agent
• What context/instructions it provides
• How it retrieves documents (optional)

​

Required Methods

​

build_context()

def build_context(self, **kwargs) -> str:
    """Return instructions about how to use this knowledge."""
    return "Use search_docs to find information."

​

get_tools()

def get_tools(self, **kwargs) -> List[Callable]:
    """Return callable tools for the agent."""
    return [self.search_docs]

​

aget_tools()

async def aget_tools(self, **kwargs) -> List[Callable]:
    """Return callable tools for the agent."""
    return [self.search_docs]

​

Optional Methods

​

retrieve()

def retrieve(self, query: str, **kwargs) -> List[Document]:
    """Retrieve documents for the query."""
    results = self._internal_search(query)
    return [Document(content=r) for r in results]

​

aretrieve()

async def aretrieve(self, query: str, **kwargs) -> List[Document]:
    """Async retrieve documents."""
    results = await self._internal_search(query)
    return [Document(content=r) for r in results]

​

Document Class

• content: The document content
• metadata: Optional metadata dictionary
• id: Optional document identifier

​

Implementation Example

from typing import List, Callable
from agno.knowledge.document import Document
from agno.knowledge.protocol import KnowledgeProtocol

class MyKnowledge:
    """Custom knowledge implementation."""
    
    def __init__(self, data_source: str):
        self.data_source = data_source
    
    def build_context(self, **kwargs) -> str:
        return "Use search_docs to find information from the knowledge base."
    
    def get_tools(self, **kwargs) -> List[Callable]:
        return [self.search_docs]
    
    async def aget_tools(self, **kwargs) -> List[Callable]:
        return [self.search_docs]
    
    def search_docs(self, query: str) -> str:
        """Search the knowledge base.
        
        Args:
            query: Search query
        """
        # Your search implementation
        results = self._search(query)
        return "\n\n".join(results)
    
    def retrieve(self, query: str, **kwargs) -> List[Document]:
        """Retrieve documents for context injection."""
        results = self._search(query)
        return [Document(content=r) for r in results]
    
    def _search(self, query: str) -> List[str]:
        # Your actual search logic
        return ["Result 1", "Result 2"]

​

Built-in Implementations

• PDFKnowledge: Load knowledge from PDF files
• WebKnowledge: Scrape websites for knowledge
• WikipediaKnowledge: Search Wikipedia
• TextKnowledge: Load from text files
• CSVKnowledge: Load from CSV files
• JSONKnowledge: Load from JSON files

​

Benefits

1. Flexible: Each implementation decides what tools to expose
2. Type-safe: Protocol typing ensures interface compliance
3. Custom naming: Not forced to use specific tool names
4. Optional features: Implement only what you need
5. Composable: Easy to combine multiple knowledge sources

​

Best Practices

1. Clear instructions: Provide clear context about available tools
2. Tool names: Use descriptive tool names that indicate their purpose
3. Error handling: Handle errors gracefully in tool implementations
4. Async support: Implement async variants for I/O operations
5. Metadata: Include useful metadata in Document objects
6. Batching: Support batch operations when possible