Integrations

• 1: Providers
• 1.1: Drasi
• 1.2: Anthropic
• 2: Loaders
• 2.1: arXiv
• 3: Splitters
• 3.1: Text
• 4: Vector Stores
• 4.1: Chroma
• 4.2: Postgres
• 4.3: Redis
• 5: Tools
• 5.1: Agents as Tools
• 5.2: MCP Toolbox for Databases

1 - Providers

1.1 - Drasi

The Drasi extension in Dapr Agents enables DurableAgent runs to be triggered by Drasi events.

Why Use Drasi?

Many systems need to react to changes produced by other systems in near real-time. Traditional polling cannot detect the absence of change, or changes that occur at an extremely high frequency without unnecessary load on source systems (even without considering network delay). Raw changes are usually not actionable, requiring custom change data capture (CDC) pipelines to process changes at scale and convert them into meaningful domain events. However, these pipelines can be expensive (if managed) or difficult to set up and maintain (if self-hosted).

For many use cases, Drasi is a viable alternative. Drasi is a CNCF Sandbox project that addresses the issues mentioned above with a simple architecture centered around detecting and reacting to changes:

• Sources to ingest changes from existing systems
• Queries allowing high-level “business conditions” to be defined across a variety of data sources, which emit events when those conditions are satisfied
• Reactions to push events to downstream consumers

Installation

• pip
• uv

pip install "dapr-agents[drasi]"

uv add dapr-agents[drasi]

Usage

from dapr_agents import AgentRunner, DurableAgent
from dapr_agents.agents.configs import AgentPubSubConfig

from dapr_agents.ext.drasi import drasi_trigger

agent = DurableAgent(
    name="InventoryAgent",
    pubsub=AgentPubSubConfig(
        pubsub_name="pubsub",           # Replace with your pub/sub component
        agent_topic="inventory-agent",  # Replace with your agent pub/sub topic
    ),
)

drasi_trigger(
    agent,
    query_id="low-stock-products",      # Replace with your Drasi query ID
)

AgentRunner().serve(agent)

API

drasi_trigger

drasi_trigger creates a static subscription to a Drasi query and allows agents to be triggered by Drasi events via Dapr pub/sub.

Parameters

DrasiOperation

DrasiOperation is an enum representing the supported Drasi change operations.

Operations

DrasiChangeEvent

DrasiChangeEvent is a Pydantic model representing a change event emitted by a query.

Attributes

Example Structure

{
    "op": "u",
    "ts_ms": 42,
    "seq": 1,
    "payload": {
        "source": {
            "queryId": "low-stock-products",
            "ts_ms": 42
        },
        "before": {"a": 1},
        "after": {"a": 2}
    }
}

Examples

See the Extension Examples for a list of working examples for the Drasi extension.

1.2 - Anthropic

The AnthropicChatClient connects Dapr Agents to Anthropic Claude models through the official Anthropic Python SDK and the Messages API. You can call it directly, or pass it as the llm of a DurableAgent.

The client supports:

• Text generation with system, user, assistant, and tool messages
• Streaming, with an optional on_chunk callback
• Image input
• Tool calling
• Structured output with Pydantic models
• Prompty files

Installation

The Anthropic SDK is included with the dapr-agents package, so no extra is required.

• pip
• uv

pip install dapr-agents

uv add dapr-agents

Configuration

The client reads the following environment variables when the matching constructor argument isn’t set:

The client also accepts a timeout argument (in seconds, or an httpx.Timeout-style dict) that defaults to 1500.

from dapr_agents import AnthropicChatClient

# Reads ANTHROPIC_API_KEY, and optionally ANTHROPIC_MODEL and ANTHROPIC_BASE_URL
llm = AnthropicChatClient()

# Or configure the client explicitly
llm = AnthropicChatClient(model="claude-sonnet-4-6", timeout=60)

Usage

Direct LLM calls

generate(...) accepts a string, a message dict, a message object, or a list of messages, and returns an LLMChatResponse:

from dapr_agents import AnthropicChatClient
from dapr_agents.types import UserMessage

llm = AnthropicChatClient()

response = llm.generate("Name a famous dog!")
print(response.get_message().content)

response = llm.generate([UserMessage("Tell me a joke about dogs.")])
print(response.get_message().content)

System messages are moved into the Messages API’s top-level system parameter automatically. Keyword arguments that generate(...) doesn’t handle itself are passed through to the Anthropic Messages API, so you can use options such as temperature, top_k, stop_sequences, or thinking:

response = llm.generate(
    "What is 27 * 453?",
    max_tokens=2048,
    thinking={"type": "enabled", "budget_tokens": 1024},
)

With a Durable Agent

Pass the client as the agent’s llm:

from dapr_agents import AnthropicChatClient, DurableAgent, tool
from dapr_agents.agents.configs import AgentMemoryConfig, AgentStateConfig
from dapr_agents.memory import ConversationDaprStateMemory
from dapr_agents.storage.daprstores.stateservice import StateStoreService
from dapr_agents.workflow.runners import AgentRunner


@tool
def get_weather(location: str) -> str:
    """Get the current weather for a location."""
    return f"{location}: 72F and sunny."


def main() -> None:
    weather_agent = DurableAgent(
        name="WeatherAgent",
        role="Weather Assistant",
        instructions=["Help users with weather information"],
        tools=[get_weather],
        llm=AnthropicChatClient(),
        memory=AgentMemoryConfig(
            store=ConversationDaprStateMemory(store_name="agent-memory")
        ),
        state=AgentStateConfig(
            store=StateStoreService(store_name="agent-workflow"),
        ),
    )

    runner = AgentRunner()
    try:
        runner.serve(weather_agent, port=8001)
    finally:
        runner.shutdown()


if __name__ == "__main__":
    main()

The agent translates its tools and the tool results in the conversation history into Anthropic’s tool_use and tool_result format for you.

Streaming

Pass stream=True to receive an iterator of LLMChatResponseChunk objects instead of a complete response:

from dapr_agents import AnthropicChatClient

llm = AnthropicChatClient()

for chunk in llm.generate("Name a famous dog!", stream=True):
    if chunk.result.content:
        print(chunk.result.content, end="", flush=True)
print()

You can also pass an on_chunk callback. It’s called with each LLMChatResponseChunk as the iterator yields it, which is useful for forwarding tokens to a UI or a log while other code consumes the stream:

from dapr_agents.types.message import LLMChatResponseChunk


def forward(chunk: LLMChatResponseChunk) -> None:
    if chunk.result.content:
        print(chunk.result.content, end="", flush=True)


stream = llm.generate("Write a haiku about durable workflows.", stream=True, on_chunk=forward)
for _ in stream:
    pass  # The request runs as the iterator is consumed

on_chunk only applies when stream=True, and it fires only while the returned iterator is being consumed. The stream produces three kinds of chunks:

• Text chunks, with the new text in chunk.result.content
• Tool call chunks, with the tool call ID and name, followed by partial JSON arguments, in chunk.result.tool_calls
• A final chunk with chunk.result.finish_reason set to the Anthropic stop reason, for example "end_turn" or "tool_use"

Each chunk’s metadata carries the provider, response id, model, and the latest usage reported by the API. When extended thinking is enabled, the thinking text and signatures are collected in the metadata under thinking_blocks, thinking_deltas, and thinking_signatures rather than being emitted as content.

Image input

Send images using OpenAI-style image_url content blocks. The client translates them into Anthropic image blocks. Both HTTP(S) URLs and base64 data URIs are supported:

import base64

from dapr_agents import AnthropicChatClient

llm = AnthropicChatClient()

with open("chart.png", "rb") as f:
    image_b64 = base64.b64encode(f.read()).decode()

response = llm.generate(
    [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "Compare these two images."},
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/png;base64,{image_b64}"},
                },
                {
                    "type": "image_url",
                    "image_url": {"url": "https://example.com/photo.jpg"},
                },
            ],
        }
    ]
)
print(response.get_message().content)

Data URIs must be base64-encoded and use one of the media types Anthropic accepts: image/jpeg, image/png, image/gif, or image/webp. Blocks that don’t meet these requirements are logged as a warning and sent unchanged. Content blocks already in Anthropic’s native format are passed through as-is.

Tool calling

Pass AgentTool objects (for example, functions decorated with @tool) in tools. They’re converted to Anthropic’s {name, description, input_schema} tool format. Any tool calls in the response are returned as standard tool_calls on the assistant message:

from dapr_agents import AnthropicChatClient, tool


@tool
def get_weather(location: str) -> str:
    """Get the current weather for a location."""
    return f"{location}: 72F and sunny."


llm = AnthropicChatClient()
response = llm.generate("What's the weather in Paris?", tools=[get_weather])

for tool_call in response.get_message().tool_calls or []:
    print(tool_call.function.name, tool_call.function.arguments)

tool_choice accepts either a string such as "auto", "any", or "none", or an Anthropic tool choice object such as {"type": "tool", "name": "get_weather"}. Tools passed as dicts must already use Anthropic’s tool format. OpenAI-style {"type": "function", "function": {...}} dicts aren’t converted.

Structured output

Pass a Pydantic model as response_format to get a validated instance back:

from pydantic import BaseModel

from dapr_agents import AnthropicChatClient


class Contact(BaseModel):
    name: str
    email: str
    demo_requested: bool


llm = AnthropicChatClient()
contact = llm.generate(
    "Extract: John Smith (john@example.com) wants a demo on Tuesday.",
    response_format=Contact,
)
print(contact.model_dump())

structured_mode selects how the schema is enforced:

Passing response_format=list[Contact] returns a generated wrapper model whose objects field holds the parsed list. If you combine response_format with stream=True, the request is still constrained, but you get a chunk iterator back and must buffer and parse the JSON yourself.

Prompty

Set the Prompty configuration type to anthropic and load the file with AnthropicChatClient.from_prompty(...):

---
name: Claude Assistant
model:
  api: chat
  configuration:
    type: anthropic
    name: claude-sonnet-4-6
  parameters:
    max_tokens: 1024
    temperature: 0.7
inputs:
  question:
    type: string
---
system:
You are a helpful assistant.

user:
{{question}}

from dapr_agents import AnthropicChatClient

llm = AnthropicChatClient.from_prompty("assistant.prompty")
response = llm.generate(input_data={"question": "What is Dapr?"})
print(response.get_message().content)

Examples

See the 01-llm-call-anthropic example in the Dapr Agents repository for runnable text completion and streaming samples.

2 - Loaders

2.1 - arXiv

The Arxiv Fetcher module in Dapr Agents provides a powerful interface to interact with the arXiv API. It is designed to help users programmatically search for, retrieve, and download scientific papers from arXiv. With advanced querying capabilities, metadata extraction, and support for downloading PDF files, the Arxiv Fetcher is ideal for researchers, developers, and teams working with academic literature.

Why Use the Arxiv Fetcher?

The Arxiv Fetcher simplifies the process of accessing research papers, offering features like:

• Automated Literature Search: Query arXiv for specific topics, keywords, or authors.
• Metadata Retrieval: Extract structured metadata, such as titles, abstracts, authors, categories, and submission dates.
• Precise Filtering: Limit search results by date ranges (e.g., retrieve the latest research in a field).
• PDF Downloading: Fetch full-text PDFs of papers for offline use.

How to Use the Arxiv Fetcher

Step 1: Install Required Modules

• pip
• uv

pip install arxiv

uv add arxiv

Step 2: Initialize the Fetcher

Set up the ArxivFetcher to begin interacting with the arXiv API.

from dapr_agents.document import ArxivFetcher

# Initialize the fetcher
fetcher = ArxivFetcher()

Step 3: Perform Searches

Basic Search by Query String

Search for papers using simple keywords. The results are returned as Document objects, each containing:

• text: The abstract of the paper.
• metadata: Structured metadata such as title, authors, categories, and submission dates.

# Search for papers related to "machine learning"
results = fetcher.search(query="machine learning", max_results=5)

# Display metadata and summaries
for doc in results:
    print(f"Title: {doc.metadata['title']}")
    print(f"Authors: {', '.join(doc.metadata['authors'])}")
    print(f"Summary: {doc.text}\n")

Advanced Querying

Refine searches using logical operators like AND, OR, and NOT or perform field-specific searches, such as by author.

Examples:

Search for papers on “agents” and “cybersecurity”:

results = fetcher.search(query="all:(agents AND cybersecurity)", max_results=10)

Exclude specific terms (e.g., “quantum” but not “computing”):

results = fetcher.search(query="all:(quantum NOT computing)", max_results=10)

Search for papers by a specific author:

results = fetcher.search(query='au:"John Doe"', max_results=10)

Filter Papers by Date

Limit search results to a specific time range, such as papers submitted in the last 24 hours.

from datetime import datetime, timedelta

# Calculate the date range
last_24_hours = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
today = datetime.now().strftime("%Y%m%d")

# Search for recent papers
recent_results = fetcher.search(
    query="all:(agents AND cybersecurity)",
    from_date=last_24_hours,
    to_date=today,
    max_results=5
)

# Display metadata
for doc in recent_results:
    print(f"Title: {doc.metadata['title']}")
    print(f"Authors: {', '.join(doc.metadata['authors'])}")
    print(f"Published: {doc.metadata['published']}")
    print(f"Summary: {doc.text}\n")

Step 4: Download PDFs

Fetch the full-text PDFs of papers for offline use. Metadata is preserved alongside the downloaded files.

import os
from pathlib import Path

# Create a directory for downloads
os.makedirs("arxiv_papers", exist_ok=True)

# Download PDFs
download_results = fetcher.search(
    query="all:(agents AND cybersecurity)",
    max_results=5,
    download=True,
    dirpath=Path("arxiv_papers")
)

for paper in download_results:
    print(f"Downloaded Paper: {paper['title']}")
    print(f"File Path: {paper['file_path']}\n")

Step 5: Extract and Process PDF Content

Use PyPDFReader from Dapr Agents to extract content from downloaded PDFs. Each page is treated as a separate Document object with metadata.

from pathlib import Path
from dapr_agents.document import PyPDFReader

reader = PyPDFReader()
docs_read = []

for paper in download_results:
    local_pdf_path = Path(paper["file_path"])
    documents = reader.load(local_pdf_path, additional_metadata=paper)
    docs_read.extend(documents)

# Verify results
print(f"Extracted {len(docs_read)} documents.")
print(f"First document text: {docs_read[0].text}")
print(f"Metadata: {docs_read[0].metadata}")

Practical Applications

The Arxiv Fetcher enables various use cases for researchers and developers:

• Literature Reviews: Quickly retrieve and organize relevant papers on a given topic or by a specific author.
• Trend Analysis: Identify the latest research in a domain by filtering for recent submissions.
• Offline Research Workflows: Download and process PDFs for local analysis and archiving.

Next Steps

While the Arxiv Fetcher provides robust functionality for retrieving and processing research papers, its output can be integrated into advanced workflows:

• Building a Searchable Knowledge Base: Combine fetched papers with integrations like text splitting and vector embeddings for advanced search capabilities.
• Retrieval-Augmented Generation (RAG): Use processed papers as inputs for RAG pipelines to power question-answering systems.
• Automated Literature Surveys: Generate summaries or insights based on the fetched and processed research.

3 - Splitters

3.1 - Text

The Text Splitter module is a foundational integration in Dapr Agents designed to preprocess documents for use in Retrieval-Augmented Generation (RAG) workflows and other in-context learning applications. Its primary purpose is to break large documents into smaller, meaningful chunks that can be embedded, indexed, and efficiently retrieved based on user queries.

By focusing on manageable chunk sizes and preserving contextual integrity through overlaps, the Text Splitter ensures documents are processed in a way that supports downstream tasks like question answering, summarization, and document retrieval.

Why Use a Text Splitter?

When building RAG pipelines, splitting text into smaller chunks serves these key purposes:

• Enabling Effective Indexing: Chunks are embedded and stored in a vector database, making them retrievable based on similarity to user queries.
• Maintaining Semantic Coherence: Overlapping chunks help retain context across splits, ensuring the system can connect related pieces of information.
• Handling Model Limitations: Many models have input size limits. Splitting ensures text fits within these constraints while remaining meaningful.

This step is crucial for preparing knowledge to be embedded into a searchable format, forming the backbone of retrieval-based workflows.

Strategies for Text Splitting

The Text Splitter supports multiple strategies to handle different types of documents effectively. These strategies balance the size of each chunk with the need to maintain context.

1. Character-Based Length

• How It Works: Counts the number of characters in each chunk.
• Use Case: Simple and effective for text splitting without dependency on external tokenization tools.

Example:

from dapr_agents.document.splitter.text import TextSplitter

# Character-based splitter (default)
splitter = TextSplitter(chunk_size=1024, chunk_overlap=200)

2. Token-Based Length

• How It Works: Counts tokens, which are the semantic units used by language models (e.g., words or subwords).
• Use Case: Ensures compatibility with models like GPT, where token limits are critical.

Example:

import tiktoken
from dapr_agents.document.splitter.text import TextSplitter

enc = tiktoken.get_encoding("cl100k_base")

def length_function(text: str) -> int:
    return len(enc.encode(text))

splitter = TextSplitter(
    chunk_size=1024,
    chunk_overlap=200,
    chunk_size_function=length_function
)

The flexibility to define the chunk size function makes the Text Splitter adaptable to various scenarios.

Chunk Overlap

To preserve context, the Text Splitter includes a chunk overlap feature. This ensures that parts of one chunk carry over into the next, helping maintain continuity when chunks are processed sequentially.

Example:

• With chunk_size=1024 and chunk_overlap=200, the last 200 tokens or characters of one chunk appear at the start of the next.
• This design helps in tasks like text generation, where maintaining context across chunks is essential.

How to Use the Text Splitter

Here’s a practical example of using the Text Splitter to process a PDF document:

Step 1: Load a PDF

import requests
from pathlib import Path

# Download PDF
pdf_url = "https://arxiv.org/pdf/2412.05265.pdf"
local_pdf_path = Path("arxiv_paper.pdf")

if not local_pdf_path.exists():
    response = requests.get(pdf_url)
    response.raise_for_status()
    with open(local_pdf_path, "wb") as pdf_file:
        pdf_file.write(response.content)

Step 2: Read the Document

For this example, we use Dapr Agents’ PyPDFReader.

• pip
• uv

pip install pypdf

uv add pypdf

Then, initialize the reader to load the PDF file.

from dapr_agents.document.reader.pdf.pypdf import PyPDFReader

reader = PyPDFReader()
documents = reader.load(local_pdf_path)

Step 3: Split the Document

import tiktoken
from dapr_agents.document.splitter.text import TextSplitter

enc = tiktoken.get_encoding("cl100k_base")

def length_function(text: str) -> int:
    return len(enc.encode(text))

splitter = TextSplitter(
    chunk_size=1024,
    chunk_overlap=200,
    chunk_size_function=length_function
)
chunked_documents = splitter.split_documents(documents)

Step 4: Analyze Results

print(f"Original document pages: {len(documents)}")
print(f"Total chunks: {len(chunked_documents)}")
print(f"First chunk: {chunked_documents[0]}")

Key Features

• Hierarchical Splitting: Splits text by separators (e.g., paragraphs), then refines chunks further if needed.
• Customizable Chunk Size: Supports character-based and token-based length functions.
• Overlap for Context: Retains portions of one chunk in the next to maintain continuity.
• Metadata Preservation: Each chunk retains metadata like page numbers and start/end indices for easier mapping.

By understanding and leveraging the Text Splitter, you can preprocess large documents effectively, ensuring they are ready for embedding, indexing, and retrieval in advanced workflows like RAG pipelines.

4 - Vector Stores

Dapr Agents includes built-in vector store implementations for use with ConversationVectorMemory and RAG pipelines. Each store is available from dapr_agents.storage.vectorstores.

Vector stores share the same interface and are interchangeable as the vector_store argument to ConversationVectorMemory:

from dapr_agents.storage.vectorstores import ChromaVectorStore  # Replace with your vector store
from dapr_agents.document.embedder.openai import OpenAIEmbedder  # Replace with your embedding model
from dapr_agents.memory import ConversationVectorMemory

store = ChromaVectorStore(
    collection_name="my_collection",
    embedding_function=OpenAIEmbedder(),
)
memory = ConversationVectorMemory(
    vector_store=store,
    distance_metric="cosine",
)

To keep the core installation minimal, vector store dependencies must be installed separately.

4.1 - Chroma

Uses ChromaDB for in-memory or persistent vector search.

Installation

• pip
• uv

pip install chromadb

uv add chromadb

Usage

from dapr_agents.storage.vectorstores import ChromaVectorStore
from dapr_agents.document.embedder.openai import OpenAIEmbedder  # Replace with your embedding model

store = ChromaVectorStore(
    collection_name="my_collection",
    embedding_function=OpenAIEmbedder(),
)

4.2 - Postgres

Uses Postgres with pgvector for production-grade vector similarity search.

Installation

• pip
• uv

pip install "psycopg[binary,pool]" pgvector

uv add 'psycopg[binary,pool]' pgvector

Usage

from dapr_agents.storage.vectorstores import PostgresVectorStore
from dapr_agents.document.embedder.openai import OpenAIEmbedder  # Replace with your embedding model

store = PostgresVectorStore(
    connection_string="postgresql://user:pass@localhost:5432/mydb",
    embedding_function=OpenAIEmbedder(),
    embedding_dimensions=1536,
)

4.3 - Redis

Uses Redis Stack via the redisvl library for vector similarity search.

Installation

• pip
• uv

pip install redisvl

uv add redisvl

Usage

from dapr_agents.storage.vectorstores import RedisVectorStore
from dapr_agents.document.embedder.openai import OpenAIEmbedder  # Replace with your embedding model

store = RedisVectorStore(
    url="redis://localhost:6379",
    index_name="my_agent",
    embedding_function=OpenAIEmbedder(),
    embedding_dimensions=1536,
    distance_metric="cosine",  # "cosine", "l2", or "ip"
    storage_type="hash",       # "hash" or "json"
)

5 - Tools

5.1 - Agents as Tools

Dapr Agents supports invoking other agents as tools within an instance of a DurableAgent reasoning loop, including agents from other frameworks such as OpenAI Agents, LangGraph, and CrewAI.

For full documentation and code examples, see Agents as Tools.

5.2 - MCP Toolbox for Databases

Dapr Agents supports integrating with MCP Toolbox for Databases by implementing a wrapper that loads the available tools into the Tool model Dapr Agents utilize.

To integrate the Toolbox, load the tools as follows:

from toolbox_core import ToolboxSyncClient

client = ToolboxSyncClient("http://127.0.0.1:5000")
try:
    agent_tools = AgentTool.from_toolbox_many(client.load_toolset("your-tools-name-here"))
    agent = DurableAgent(
        # ...
        tools=agent_tools,
    )
finally:
    client.close()

Or wrap it in a with statement:

from toolbox_core import ToolboxSyncClient
with ToolboxSyncClient("http://127.0.0.1:5000") as client:
    agent_tools = AgentTool.from_toolbox_many(client.load_toolset("your-tools-name-here"))
    agent = DurableAgent(
        # ...
        tools=agent_tools,
    )