How to Create Custom Sources

Custom sources let you extend the built-in agent with new capabilities — a new tool set, a new skill provider, a new context source — without writing a custom agent class. Once registered, a custom source is available to any agent configuration via include_sources.

Scaffold a Custom Source

The CLI scaffolds source files for any provider:

za agents capabilities <provider> add <source-name>

Provider CLI name	Source base class	What it provides
tool-sets	ToolsSource	Tools the LLM can call
skill-sources	SkillsSource	Skills injected into the prompt
context-sources	ContextSource	Context injected before the conversation
memory-stores	MemoryStore	Persistent memory read/write
instruction-sources	InstructionSource	System prompt instructions
delegation-sources	DelegableAgentsSource	Named agents for delegation
processors	MessageProcessor	Post-processing of agent responses
dispatch-rules	DispatchRule	Routing rules for multi-agent dispatch

For example, to create a custom tools source:

za agents capabilities tool-sets add my-custom-tools

This generates dependencies/my_custom_tools.py with the boilerplate:

from zav.pydantic_compat import BaseModel, Field
from zav.agents_sdk import AgentDependencyFactory, AgentDependencyRegistry
from zav.agents_sdk.adapters.tools.tools_provider import ToolsSource


class MyCustomToolsConfiguration(BaseModel):
    enabled: bool = Field(False, description="Whether this source is enabled.")


class MyCustomToolsSource(ToolsSource):
    source_name = "my_custom_tools"

    def __init__(self, my_custom_tools_configuration: MyCustomToolsConfiguration):
        self.enabled = my_custom_tools_configuration.enabled


class MyCustomToolsSourceFactory(AgentDependencyFactory):
    @classmethod
    def create(
        cls,
        my_custom_tools_configuration: MyCustomToolsConfiguration = MyCustomToolsConfiguration(),
    ) -> MyCustomToolsSource:
        return MyCustomToolsSource(my_custom_tools_configuration=my_custom_tools_configuration)


AgentDependencyRegistry.register(MyCustomToolsSourceFactory)

Implement the Source

Fill in the source class with your logic. Each source base class has its own interface — implement the required methods:

from typing import Dict, Any, List
from zav.agents_sdk.domain.tools import Tool, streamable, hide
from zav.agents_sdk.adapters.tools.tools_provider import ToolsSource


class MyCustomToolsSource(ToolsSource):
    source_name = "my_custom_tools"

    def __init__(self, my_custom_tools_configuration: MyCustomToolsConfiguration):
        self.enabled = my_custom_tools_configuration.enabled
        self.config = my_custom_tools_configuration

    async def get_tools(self) -> List[Tool]:
        return [
            Tool.from_callable(
                name="search_documents",
                executable=self.search_documents,
            ),
        ]

    @streamable(
        running_text="Searching for '{{ query }}'...",
        completed_text="Found {{ result_count }} result{{ 's' if result_count != 1 else '' }}.",
        params_transform=hide,
        response_transform=hide,
    )
    async def search_documents(self, query: str) -> Dict[str, Any]:
        """Search for documents matching the query."""
        results = await self.__perform_search(query)
        return {"result_count": len(results), "documents": results}

Processors: stream-based interface

Unlike other sources that return collections, MessageProcessor uses a generator-chaining pattern. Each processor wraps the previous stream and yields transformed (ChatResponse, ChatMessage) pairs:

from typing import AsyncGenerator
from zav.agents_sdk.adapters.message_processing.message_processor import (
    MessageProcessor, StreamItem,
)


class MyProcessorSource(MessageProcessor):
    source_name = "my_processor"

    async def process_stream(
        self, stream: AsyncGenerator[StreamItem, None]
    ) -> AsyncGenerator[StreamItem, None]:
        async for response, message in stream:
            # Transform the message before yielding
            yield response, message

Enable the Source

Once registered, enable your source in any agent's configuration:

za agents capabilities tool-sets configure my-custom-tools

Or add it directly to agent_setups.json:

{
  "tools_provider_configuration": {
    "include_sources": ["my_custom_tools"]
  },
  "my_custom_tools_source_configuration": {
    "enabled": true
  }
}

How It Works

The SDK's provider-source pattern makes this seamless:

1. Your source factory is registered with AgentDependencyRegistry
2. When the agent is built, DependencyGroup auto-collects all registered sources of the matching type
3. The provider applies source filtering (include_sources / exclude_sources) from configuration
4. Active sources contribute their capabilities to the agent

This means your source is immediately composable — it can be combined with any other source, enabled or disabled per-agent, and filtered at any configuration layer (platform, tenant, agent).