Dapr Agents

The MCPServer specification allows wiring workflow-based middleware hooks around tool calls. These hooks enable cross-cutting concerns like security and observability to be implemented as reusable workflows.

Hook Types

• beforeCallTool: Executed before the tool is invoked. If a beforeCallTool hook returns an error, the tool call is aborted.
• afterCallTool: Executed after the tool invocation. These hooks can log errors or results without affecting the actual tool output.

Common Middleware Implementations

Multi-agent systems in Dapr Agents rely on a shared registry and specific messaging patterns for coordination.

1. Shared Registry

Agents use AgentRegistryConfig with the same team_name to discover each other. This allows orchestrators to query the registry for available members.

registry = AgentRegistryConfig(
    store=StateStoreService(store_name="agentregistrystore"),
    team_name="fellowship",  # Must be identical for all agents in the team
)

2. Agent Discovery

An orchestrator can find all members of a team using the registry:

# Returns a list of available agents, e.g., [frodo, sam, gandalf, legolas]
available_agents = registry.get_team_members("fellowship")

3. Message Flow Patterns

• Via Orchestrator: Client $\rightarrow$ Orchestrator Topic $\rightarrow$ Orchestrator Logic $\rightarrow$ Agent Topic $\rightarrow$ Specific Agent
• Direct to Agent: Client $\rightarrow$ Agent Topic $\rightarrow$ Specific Agent
• Broadcast: All agents subscribed to a team-wide broadcast topic (e.g., fellowship.broadcast) receive the message simultaneously.

The otel-configstore agent variant supports real-time OTel configuration updates by subscribing to the Dapr Configuration Store. This allows you to change observability settings without restarting the agent pods.

Supported Configuration Keys

Example: Runtime Updates

To update configuration at runtime, use redis-cli to set the keys in the Redis instance used by the Dapr Configuration Store.

NAMESPACE=dapr-agents
REDIS_POD=$(kubectl get pods -n $NAMESPACE -l app.kubernetes.io/name=redis -o jsonpath='{.items[0].metadata.name}')

# Switch to a new collector endpoint
kubectl exec $REDIS_POD -n $NAMESPACE -- redis-cli SET otel_exporter_otlp_endpoint "http://new-collector:4317"

# Disable OTel
kubectl exec $REDIS_POD -n $NAMESPACE -- redis-cli SET otel_sdk_disabled "true"

# Re-enable with a different exporter
kubectl exec $REDIS_POD -n $NAMESPACE -- redis-cli SET otel_sdk_disabled "false"
kubectl exec $REDIS_POD -n $NAMESPACE -- redis-cli SET otel_traces_exporter "otlp_http"

The @message_router decorator is applied directly to a workflow function to turn it into a Pub/Sub handler. This eliminates the need for a separate handler function and manual workflow scheduling.

When a message arrives on the specified topic:

1. The workflow is automatically scheduled.
2. The incoming message is automatically validated against the provided message_model (e.g., a Pydantic model).
3. Optional hooks (payload_filter, model_filter, mapper) are executed to decide if the workflow should run and how the input should be shaped.

Important Constraints:

• Blocking Nature: Hooks run on the per-topic consumer thread. They must be fast, in-memory, and side-effect-free. Do not perform external I/O (like database lookups or HTTP calls) inside hooks; instead, perform those checks inside the workflow body.
• No Async Hooks: async def hooks are rejected at decoration time because they would still block the consumer thread.

@message_router(
    pubsub="messagepubsub",
    topic="blog.requests",
    message_model=StartBlogMessage,
)
def my_workflow(ctx, wf_input: dict) -> str:
    ...

The DaprChatClient is the primary interface for interacting with Large Language Models (LLMs) via Dapr's Conversation API. It allows you to write provider-agnostic code that can switch between different LLM backends (like OpenAI, Anthropic, or local echo components) simply by changing Dapr component configurations, without modifying your application logic.

Key capabilities include:

• Provider Agnosticism: Switch LLM providers via Dapr components.
• Prompt Caching: Reduce latency and costs.
• PII Obfuscation: Mask sensitive information.
• Resilience: Built-in support for retries, timeouts, and circuit breaking via Dapr resiliency policies.

from dapr_agents.llm import DaprChatClient
from dapr_agents.types import UserMessage

# Basic usage
llm = DaprChatClient()
response = llm.generate("Name a famous dog!")

# Using a prompty file for context
llm = DaprChatClient.from_prompty("basic.prompty")
response = llm.generate(input_data={"question": "What is your name?"})

# Using explicit messages
response = llm.generate(messages=[UserMessage("hello")])

Dapr Agents allows one DurableAgent to call another agent as a synchronous child workflow tool. When an LLM selects an agent as a tool, the framework schedules the target agent's workflow as a Dapr child workflow. The result of the target agent's execution is returned to the caller as a ToolMessage, integrating seamlessly into the caller's conversation history and durable execution model.

There are three primary ways to configure an agent to act as a tool for another agent:

1. Explicit Cross-App Factory: Use agent_to_tool to wire agents that live in separate Dapr applications. This method does not require a shared registry.
2. Direct Instance Passing (In-Process): Pass a DurableAgent instance directly into the tools list of another agent. This is used when both agents run in the same Dapr app/Python process; the framework auto-converts the instance into an AgentWorkflowTool.
3. Shared Registry (Auto-discovery): If both agents are configured with a registry, the framework automatically discovers peer agents at the start of a workflow run. The load_tools mechanism scans the registry and registers all peers (excluding orchestrators and self) in the tool_executor.

from dapr_agents.tool.workflow.agent_tool import agent_to_tool

# 1. Explicit cross-app factory (no registry needed)
sam_tool = agent_to_tool(
    "sam",
    description="Sam Gamgee. Goal: Manage provisions.",
    target_app_id="SamApp",
)
frodo = DurableAgent(name="frodo", tools=[sam_tool], ...)

# 2. Pass DurableAgent instance directly (auto-converted, same app)
sam = DurableAgent(name="sam", registry=registry, ...)
frodo = DurableAgent(name="frodo", tools=[sam], ...)

# 3. Shared registry (auto-discovery)
sam = DurableAgent(name="sam", registry=registry, ...)
frodo = DurableAgent(name="frodo", registry=registry, ...)