cadence-sdk 2.0.5

Framework-agnostic SDK for building Cadence AI agent plugins

Cadence SDK

Framework-agnostic plugin development kit for multi-tenant AI agent platforms

Cadence SDK is a Python library that enables developers to build AI agent plugins that work seamlessly across multiple orchestration frameworks (LangGraph, OpenAI Agents SDK, Google ADK) without framework-specific code.

Table of Contents

• Features
• Installation
• Quick Start
• Core Concepts
• Plugin Development
• Tool Development
• Caching
• State Management
• Plugin Registry
• Plugin Discovery
• Dependency Management
• Validation
• Examples
• API Reference
• Best Practices
• Contributing

Features

Framework-Agnostic Design

Write your plugin once, run it on any supported orchestration framework:

• LangGraph (LangChain-based)
• OpenAI Agents SDK
• Google ADK (Agent Development Kit)

Simple Tool Declaration

Define tools with a single decorator — no framework-specific code:

@uvtool
def search(query: str) -> str:
    """Search for information."""
    return perform_search(query)

Integrated Caching

Built-in semantic caching for expensive operations:

@uvtool(cache=CacheConfig(ttl=3600, similarity_threshold=0.85))
def expensive_api_call(query: str) -> str:
    """Cached API call."""
    return call_external_api(query)

Plugin System

• Plugin discovery from multiple sources (pip packages, directories, system-wide)
• Settings schema with type validation
• Dependency management with install helpers
• Health checks and lifecycle management
• Plugin registry with version support and capability filtering

Type Safety

Fully typed with Pydantic for excellent IDE support and runtime validation.

Async Support

First-class support for async tools with automatic detection and invocation.

Installation

From PyPI

pip install cadence-sdk

From Source

git clone https://github.com/jonaskahn/cadence-sdk.git
cd cadence-sdk
poetry install

Development Installation

poetry install --with dev

Quick Start

1. Create Your First Plugin

# my_plugin/plugin.py
from cadence_sdk import (
    BasePlugin, BaseAgent, PluginMetadata,
    uvtool, UvTool, plugin_settings, CacheConfig
)
from typing import List


class MyAgent(BaseAgent):
    """My custom agent."""

    def __init__(self):
        self.greeting = "Hello"
        self._greet_tool = self._make_greet_tool()
        self._search_tool = self._make_search_tool()

    def initialize(self, config: dict) -> None:
        self.greeting = config.get("greeting", "Hello")

    def _make_greet_tool(self) -> UvTool:
        @uvtool
        def greet(name: str) -> str:
            """Greet a user by name."""
            return f"{self.greeting}, {name}!"

        return greet

    def _make_search_tool(self) -> UvTool:
        @uvtool(cache=CacheConfig(ttl=3600, similarity_threshold=0.85))
        def search(query: str) -> str:
            """Search for information (cached)."""
            return f"Results for: {query}"

        return search

    def get_tools(self) -> List[UvTool]:
        return [self._greet_tool, self._search_tool]

    def get_system_prompt(self) -> str:
        return "You are a helpful assistant."


@plugin_settings([
    {
        "key": "api_key",
        "name": "API Key",
        "type": "str",
        "required": True,
        "sensitive": True,
        "description": "API key for external service"
    },
    {
        "key": "greeting",
        "name": "Greeting",
        "type": "str",
        "default": "Hello",
        "description": "Greeting phrase"
    }
])
class MyPlugin(BasePlugin):
    """My custom plugin."""

    @staticmethod
    def get_metadata() -> PluginMetadata:
        return PluginMetadata(
            pid="com.example.my_plugin",
            name="My Plugin",
            version="1.0.0",
            description="My awesome plugin",
            capabilities=["greeting", "search"],
        )

    @staticmethod
    def create_agent() -> BaseAgent:
        return MyAgent()

2. Register Your Plugin

from cadence_sdk import register_plugin
from my_plugin import MyPlugin

contract = register_plugin(MyPlugin)

3. Use Your Plugin

Your plugin is now ready to be loaded by the Cadence platform and will work with any supported orchestration framework!

Core Concepts

Plugins

Plugins are factory classes that create agent instances. They declare metadata, settings schema, and provide health checks. The pid (plugin ID) is a required reverse-domain identifier (e.g., com.example.my_plugin) used as the registry key.

class MyPlugin(BasePlugin):
    @staticmethod
    def get_metadata() -> PluginMetadata:
        return PluginMetadata(
            pid="com.example.my_plugin",
            name="My Plugin",
            version="1.0.0",
            description="Description",
            capabilities=["cap1", "cap2"],
            dependencies=["requests>=2.0"],
        )

    @staticmethod
    def create_agent() -> BaseAgent:
        return MyAgent()

    @staticmethod
    def validate_dependencies() -> List[str]:
        """Return list of error strings, empty if all deps are satisfied."""
        return []

    @staticmethod
    def health_check() -> dict:
        return {"status": "healthy"}

Agents

Agents provide tools and system prompts. They can maintain state and be initialized with configuration.

Because tools often need access to agent state, the recommended pattern is to create tools inside methods using closures:

class MyAgent(BaseAgent):
    def __init__(self):
        self.api_key = None
        self._search_tool = self._make_search_tool()

    def initialize(self, config: dict) -> None:
        """Initialize with configuration."""
        self.api_key = config.get("api_key")

    def _make_search_tool(self) -> UvTool:
        @uvtool
        def search(query: str) -> str:
            """Search using the configured API key."""
            return call_api(query, self.api_key)

        return search

    def get_tools(self) -> List[UvTool]:
        return [self._search_tool]

    def get_system_prompt(self) -> str:
        return "You are a helpful assistant."

    async def cleanup(self) -> None:
        """Clean up resources."""
        pass

Tools

Tools are functions that agents can invoke. They can be synchronous or asynchronous.

from cadence_sdk import uvtool, CacheConfig
from pydantic import BaseModel


# Simple tool
@uvtool
def simple_tool(text: str) -> str:
    """A simple tool."""
    return text.upper()


# Tool with args schema
class SearchArgs(BaseModel):
    query: str
    limit: int = 10


@uvtool(args_schema=SearchArgs)
def search(query: str, limit: int = 10) -> str:
    """Search with validation."""
    return f"Top {limit} results for: {query}"


# Cached tool
@uvtool(cache=CacheConfig(
    ttl=3600,
    similarity_threshold=0.85,
    cache_key_fields=["query"]  # Only cache by query
))
def expensive_search(query: str, options: dict = None) -> str:
    """Expensive operation with selective caching."""
    return perform_expensive_search(query, options)


# Async tool
@uvtool
async def async_fetch(url: str) -> str:
    """Asynchronous tool."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

Messages

Framework-agnostic message types for agent communication:

from cadence_sdk import (
    UvHumanMessage,
    UvAIMessage,
    UvSystemMessage,
    UvToolMessage,
    ToolCall
)

# Human message
human = UvHumanMessage(content="Hello!")

# AI message with tool calls
ai = UvAIMessage(
    content="Let me search for that.",
    tool_calls=[
        ToolCall(name="search", args={"query": "Python"})
    ]
)

# System message
system = UvSystemMessage(content="You are helpful.")

# Tool result message
tool_result = UvToolMessage(
    content="Search results: ...",
    tool_call_id="call_123",
    tool_name="search"
)

State

The UvState TypedDict provides a minimal universal state structure used across frameworks:

from cadence_sdk import UvState, UvHumanMessage

state: UvState = {
    "messages": [UvHumanMessage(content="Hello")],
    "thread_id": "thread_123",
}

Plugin Development

Settings Declaration

Declare settings schema for your plugin using @plugin_settings:

from cadence_sdk import plugin_settings


@plugin_settings([
    {
        "key": "api_key",
        "name": "API Key",  # Display name shown in UI
        "type": "str",
        "required": True,
        "sensitive": True,  # Masks value in logs/UI
        "description": "API key for service"
    },
    {
        "key": "max_results",
        "name": "Max Results",
        "type": "int",
        "default": 10,
        "required": False,
        "description": "Maximum results to return"
    },
    {
        "key": "endpoints",
        "name": "Endpoints",
        "type": "list",
        "default": ["https://api.example.com"],
        "description": "API endpoints"
    }
])
class MyPlugin(BasePlugin):
    pass

Setting field types: "str", "int", "float", "bool", "list", "dict"

Agent Initialization

Agents receive resolved settings during initialization:

class MyAgent(BaseAgent):
    def __init__(self):
        self.api_key = None
        self.max_results = 10

    def initialize(self, config: dict) -> None:
        """Initialize with resolved configuration.

        Config contains:
        - Declared settings with defaults applied
        - User-provided overrides
        - Framework-resolved values
        """
        self.api_key = config["api_key"]
        self.max_results = config.get("max_results", 10)

Resource Cleanup

Implement cleanup for proper resource management:

class MyAgent(BaseAgent):
    def __init__(self):
        self.db_connection = None
        self.http_client = None

    async def cleanup(self) -> None:
        """Clean up resources when agent is disposed."""
        if self.db_connection:
            await self.db_connection.close()

        if self.http_client:
            await self.http_client.aclose()

Tool Development

Basic Tool

@uvtool
def greet(name: str) -> str:
    """Greet a user by name.

    Args:
        name: Name of the person to greet

    Returns:
        Greeting message
    """
    return f"Hello, {name}!"

Tool with Schema Validation

from pydantic import BaseModel, Field


class SearchArgs(BaseModel):
    query: str = Field(..., description="Search query")
    limit: int = Field(10, ge=1, le=100, description="Max results")
    filters: dict = Field(default_factory=dict, description="Search filters")


@uvtool(args_schema=SearchArgs)
def search(query: str, limit: int = 10, filters: dict = None) -> str:
    """Search with validated arguments."""
    return perform_search(query, limit, filters or {})

Async Tool

@uvtool
async def fetch_data(url: str) -> dict:
    """Asynchronously fetch data from URL.

    The SDK automatically detects async functions and handles
    invocation correctly.
    """
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()


# Invoke async tool
result = await fetch_data.ainvoke(url="https://api.example.com")

Tool Invocation

# Sync tool - direct call
result = greet(name="Alice")

# Sync tool - explicit invoke
result = greet.invoke(name="Alice")

# Async tool - must use ainvoke
result = await fetch_data.ainvoke(url="https://example.com")

# Check if tool is async
if fetch_data.is_async:
    result = await fetch_data.ainvoke(url="https://example.com")
else:
    result = fetch_data(url="https://example.com")

Caching

Cache Configuration

from cadence_sdk import uvtool, CacheConfig


# Method 1: CacheConfig instance (recommended)
@uvtool(cache=CacheConfig(
    ttl=3600,  # Cache for 1 hour
    similarity_threshold=0.85,  # 85% similarity for cache hits
    cache_key_fields=["query"]  # Only cache by query parameter
))
def cached_search(query: str, limit: int = 10) -> str:
    """Different limits use same cached result."""
    return expensive_search(query, limit)


# Method 2: Dictionary
@uvtool(cache={
    "ttl": 7200,
    "similarity_threshold": 0.9
})
def another_cached_tool(text: str) -> str:
    return process(text)


# Method 3: Boolean (use defaults)
@uvtool(cache=True)  # TTL=3600, threshold=0.85
def simple_cached_tool(input: str) -> str:
    return expensive_operation(input)


# Disable caching
@uvtool(cache=False)
# or simply:
@uvtool
def no_cache_tool(data: str) -> str:
    return realtime_data()

Cache Configuration Options

Field	Type	Default	Description
enabled	bool	True	Whether caching is enabled
ttl	int	3600	Time-to-live in seconds
similarity_threshold	float	0.85	Cosine similarity threshold (0.0-1.0)
cache_key_fields	List[str]	None	Fields to use for cache key (None = all)

How Caching Works

1. Semantic Matching: Uses embeddings to find similar queries
2. Threshold: Only returns cached results above similarity threshold
3. TTL: Cached results expire after TTL seconds
4. Selective Keys: Cache only by specific parameters

@uvtool(cache=CacheConfig(
    ttl=3600,
    similarity_threshold=0.85,
    cache_key_fields=["query"]
))
def search(query: str, limit: int = 10, format: str = "json") -> str:
    """Cache by query only, ignore limit and format."""
    pass


# These will use the same cached result:
search("Python programming", limit=10, format="json")
search("Python programming", limit=50, format="xml")

# This might get a cache hit if similarity > 0.85:
search("Python coding", limit=10, format="json")

State Management

The UvState TypedDict provides a framework-agnostic conversation state:

from cadence_sdk import UvState, UvHumanMessage

# UvState fields:
#   messages: List[AnyMessage]  - conversation messages
#   thread_id: Optional[str]    - conversation thread identifier

state: UvState = {
    "messages": [UvHumanMessage(content="Hello")],
    "thread_id": "thread_123",
}

State management beyond this (routing history, agent hops, plugin context) is handled by the Cadence platform internally and is not exposed through the SDK.

Plugin Registry

The PluginRegistry is a singleton that manages all registered plugins:

from cadence_sdk import PluginRegistry, register_plugin

# Register a plugin (convenience function)
contract = register_plugin(MyPlugin)

# Or use the registry directly
registry = PluginRegistry.instance()
registry.register(MyPlugin)

# Lookup plugins
contract = registry.get_plugin("com.example.my_plugin")
contract = registry.get_plugin_by_version("com.example.my_plugin", "1.0.0")

# List plugins
all_plugins = registry.list_registered_plugins()
search_plugins = registry.list_plugins_by_capability("search")
specialized = registry.list_plugins_by_type("specialized")
versions = registry.list_plugin_versions("com.example.my_plugin")

# Check existence
if registry.has_plugin("com.example.my_plugin"):
    ...

# Unregister
registry.unregister("com.example.my_plugin")

PluginContract

register_plugin returns a PluginContract that provides a standardized interface:

contract = register_plugin(MyPlugin)

print(contract.pid)  # "com.example.my_plugin"
print(contract.name)  # "My Plugin"
print(contract.version)  # "1.0.0"
print(contract.description)  # "My awesome plugin"
print(contract.capabilities)  # ["search", "greeting"]
print(contract.is_stateless)  # True

agent = contract.create_agent()
errors = contract.validate_dependencies()
health = contract.health_check()

Plugin Discovery

Discover plugins from the filesystem automatically:

from cadence_sdk import discover_plugins, DirectoryPluginDiscovery

# Convenience function
plugins = discover_plugins(["/path/to/plugins", "/another/path"])
for p in plugins:
    print(f"Found: {p.name} v{p.version}")

# Class-based (more control)
discovery = DirectoryPluginDiscovery(
    search_paths=["/path/to/plugins"],
    auto_register=True  # Auto-register with PluginRegistry
)
plugins = discovery.discover()
discovered = discovery.get_discovered()  # Dict[pid, PluginContract]

discovery.reset()  # Re-scan directories

Dependency Management

from cadence_sdk import install_dependencies, check_dependency_installed

# Check if a package is installed
if not check_dependency_installed("requests"):
    success, message = install_dependencies(["requests>=2.28"])
    if not success:
        print(f"Installation failed: {message}")

# Install multiple packages
success, output = install_dependencies(
    ["aiohttp>=3.8", "pydantic>=2.0"],
    upgrade=False,
    quiet=True
)

Additional helpers available via direct import from cadence_sdk.utils:

from cadence_sdk.utils import (
    install_plugin_dependencies,  # Install for a specific plugin
    get_installed_version,  # Get installed package version
    extract_package_name,  # Extract name from "requests>=2.28"
)

Validation

from cadence_sdk import validate_plugin_structure, validate_plugin_structure_shallow

# Shallow validation (fast, no instantiation)
is_valid, errors = validate_plugin_structure_shallow(MyPlugin)

# Deep validation (instantiates agent, checks tools, validates SDK version)
is_valid, errors = validate_plugin_structure(MyPlugin)

if not is_valid:
    for error in errors:
        print(f"ERROR: {error}")

Both functions return Tuple[bool, List[str]] — a validity flag and a list of error messages.

Deep validation checks:

• Is it a BasePlugin subclass with required methods?
• Can an agent be created?
• Are all tools valid UvTool instances?
• Is the system prompt non-empty?
• Is the SDK version compatible?
• Are declared dependencies satisfiable?

Examples

Complete Plugin Example

See the template_plugin for a complete, working example that demonstrates:

• Plugin and agent structure
• Sync and async tools
• Caching configuration
• Settings schema
• Resource cleanup

Running the Example

cd cadence-sdk
PYTHONPATH=src python examples/test_sdk.py

Running the Test Suite

cd cadence-sdk
poetry install --with dev
PYTHONPATH=src python -m pytest tests/ -v

# With coverage
PYTHONPATH=src python -m pytest tests/ --cov=cadence_sdk --cov-report=term-missing

API Reference

Core Classes

BasePlugin

Abstract base class for plugins.

Method	Signature	Description
get_metadata	() -> PluginMetadata	Return plugin metadata (required)
create_agent	() -> BaseAgent	Create agent instance (required)
validate_dependencies	() -> List[str]	Return error list, empty if OK
health_check	() -> dict	Perform health check

BaseAgent

Abstract base class for agents.

Method	Signature	Description
get_tools	() -> List[UvTool]	Return list of tools (required)
get_system_prompt	() -> str	Return system prompt (required)
initialize	(config: dict) -> None	Initialize with config (optional)
cleanup	() -> None	Async cleanup of resources (optional)

PluginMetadata

Dataclass describing plugin capabilities and requirements.

Field	Type	Default	Description
pid	str	—	Globally unique reverse-domain ID (required)
name	str	—	Human-readable display name (required)
version	str	—	Semantic version string (required)
description	str	—	Human-readable description (required)
capabilities	List[str]	[]	Capability tags
dependencies	List[str]	[]	Pip package requirements
agent_type	str	"specialized"	Agent type category
sdk_version	str	">=2.0.0,<3.0.0"	Compatible SDK version range
stateless	bool	True	Whether plugin instance can be shared

UvTool

Framework-agnostic tool wrapper.

Attribute	Type	Description
name	str	Tool name
description	str	Tool description
func	Callable	Underlying callable
args_schema	Optional[Type[BaseModel]]	Pydantic model for arguments
cache	Optional[CacheConfig]	Cache configuration
metadata	Dict[str, Any]	Additional metadata
is_async	bool	Whether tool is async

Method	Description
__call__(*args, **kwargs)	Sync invocation
invoke(*args, **kwargs)	Sync invocation alias
ainvoke(*args, **kwargs)	Async invocation

CacheConfig

Cache configuration dataclass.

Field	Type	Default	Description
enabled	bool	True	Whether caching is enabled
ttl	int	3600	Time-to-live in seconds
similarity_threshold	float	0.85	Similarity threshold (0.0-1.0)
cache_key_fields	Optional[List[str]]	None	Fields for cache key (None = all)

Loggable

Mixin that provides standardized logging for plugin classes.

class MyAgent(BaseAgent, Loggable):
    def some_method(self):
        self.logger.info("Processing...")
        self.set_log_level(logging.DEBUG)

Message Types

Class	role	Description
UvHumanMessage	"human"	Message from a human user
UvAIMessage	"ai"	Message from an AI agent (supports tool_calls)
UvSystemMessage	"system"	System instruction message
UvToolMessage	"tool"	Tool execution result
ToolCall	—	Tool invocation record (id, name, args)

Decorators

@uvtool

Convert a function to a UvTool.

Parameter	Type	Default	Description
name	str	function name	Tool name
description	str	docstring	Tool description
args_schema	Type[BaseModel]	None	Pydantic model for validation
cache	CacheConfig | bool | dict	None	Cache configuration
**metadata	any	—	Additional metadata

@plugin_settings

Declare plugin configuration schema.

Parameter	Type	Description
settings	List[dict]	List of setting definitions

Setting definition fields:

Field	Type	Required	Description
key	str	Yes	Machine-readable identifier
type	str	Yes	One of: "str", "int", "float", "bool", "list", "dict"
description	str	Yes	Human-readable description
name	str	No	Display name shown in UI (defaults to key)
default	Any	No	Default value if not provided
required	bool	No	Whether setting is mandatory (default: False)
sensitive	bool	No	Mask value in logs/UI (default: False)

Utility Functions

Function	Signature	Description
register_plugin	(plugin_class, override=False) -> PluginContract	Register plugin with global registry
discover_plugins	(search_paths, auto_register=True) -> List[PluginContract]	Discover plugins in directories
validate_plugin_structure	(plugin_class) -> Tuple[bool, List[str]]	Deep validation with instantiation
validate_plugin_structure_shallow	(plugin_class) -> Tuple[bool, List[str]]	Fast structural validation
install_dependencies	(packages, upgrade=False, quiet=True) -> Tuple[bool, str]	Install pip packages
check_dependency_installed	(package_name) -> bool	Check if package is installed

Best Practices

1. Keep Plugins Stateless

When possible, design plugins to be stateless (stateless=True in metadata). This allows the framework to share plugin instances across multiple orchestrators for better memory efficiency.

PluginMetadata(
    pid="com.example.my_plugin",
    name="My Plugin",
    version="1.0.0",
    description="Plugin description",
    stateless=True,
)

2. Create Tools as Closures for State Access

Use factory methods that return closures so tools can access agent state without globals:

class MyAgent(BaseAgent):
    def __init__(self):
        self.api_key = None
        self._search_tool = self._make_search_tool()

    def _make_search_tool(self) -> UvTool:
        @uvtool
        def search(query: str) -> str:
            """Search using agent's configured API key."""
            return call_api(query, self.api_key)  # captures self

        return search

3. Use Type Hints

Always use type hints for better IDE support and runtime validation:

@uvtool
def my_tool(query: str, limit: int = 10) -> str:
    """Type hints improve IDE support."""
    return search(query, limit)

4. Provide Good Descriptions

Tools and plugins should have clear, concise descriptions that the LLM uses for routing:

@uvtool
def search(query: str) -> str:
    """Search for information using the external API.

    This tool performs semantic search across our knowledge base
    and returns the top matching results.

    Args:
        query: The search query string

    Returns:
        Formatted search results
    """
    return perform_search(query)

5. Handle Errors Gracefully

@uvtool
def api_call(endpoint: str) -> str:
    """Make API call with proper error handling."""
    try:
        response = requests.get(endpoint)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        return f"Error: {str(e)}"

6. Use Selective Caching

Only cache by parameters that meaningfully affect the result:

@uvtool(cache=CacheConfig(
    cache_key_fields=["query", "language"],  # Ignore format, limit
))
def translate(query: str, language: str, format: str = "text", limit: int = 100) -> str:
    """Cache by query and language only."""
    pass

7. Clean Up Resources

Always implement cleanup for connections and external resources:

class MyAgent(BaseAgent):
    async def cleanup(self) -> None:
        """Clean up connections and resources."""
        if hasattr(self, "db"):
            await self.db.close()
        if hasattr(self, "http_client"):
            await self.http_client.aclose()

8. Version Your Plugins

Use semantic versioning and declare dependencies explicitly:

PluginMetadata(
    pid="com.example.my_plugin",
    name="My Plugin",
    version="1.2.3",
    description="Plugin description",
    sdk_version=">=2.0.0,<3.0.0",
    dependencies=["requests>=2.28.0", "aiohttp>=3.8.0"],
)

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

Development Setup

git clone https://github.com/jonaskahn/cadence-sdk.git
cd cadence-sdk
poetry install --with dev

# Run tests
PYTHONPATH=src python -m pytest tests/

# Run linting
poetry run ruff check .
poetry run ruff format .

Running Tests

# All tests
PYTHONPATH=src python -m pytest tests/

# With coverage
PYTHONPATH=src python -m pytest tests/ --cov=cadence_sdk --cov-report=term-missing

# Specific test file
PYTHONPATH=src python -m pytest tests/test_sdk_tools.py -v

# Run example script
PYTHONPATH=src python examples/test_sdk.py

License

MIT License — see LICENSE file for details.

Support

• Issues: GitHub Issues
• Discussions: GitHub Discussions

Changelog

See CHANGELOG.md for version history and release notes.

---

Built with for the AI agent development community