dapr-integration

Name: dapr-integration
Author: mjunaidca

26stars11forks

How to install dapr-integration?

mjunaidca/mjs-agent-skills/dapr-integration

$npx -y skills add mjunaidca/mjs-agent-skills --skill dapr-integration

Installs into the current project.

Summary

Integrate Dapr for pub/sub messaging and scheduled jobs. Use this skill when implementing event-driven architectures with Dapr, handling CloudEvent message formats, setting up pub/sub subscriptions, or scheduling jobs with Dapr Jobs API. Covers common pitfalls like CloudEvent unw

Files · 1

View on GitHub

SKILL.md

1	# Dapr Integration
2
3	Integrate Dapr sidecar for pub/sub messaging, state management, and scheduled jobs in Kubernetes environments.
4
5	## When to Use
6
7	- Setting up Dapr pub/sub for event-driven microservices
8	- Scheduling jobs with Dapr Jobs API (v1.0-alpha1)
9	- Handling CloudEvent message formats
10	- Implementing subscription handlers in FastAPI
11	- Debugging Dapr integration issues
12
13	## Quick Start
14
15	```bash
16	# Install Dapr CLI
17	curl -fsSL https://raw.githubusercontent.com/dapr/cli/master/install/install.sh \| bash
18
19	# Initialize Dapr (local)
20	dapr init
21
22	# Run app with Dapr sidecar
23	dapr run --app-id myapp --app-port 8000 -- uvicorn main:app
24	```
25
26	## Core Patterns
27
28	### 1. Pub/Sub Subscription Handler (FastAPI)
29
30	```python
31	from fastapi import APIRouter, Request
32	from sqlmodel.ext.asyncio.session import AsyncSession
33
34	router = APIRouter(prefix="/dapr", tags=["Dapr"])
35
36	# Topics we subscribe to
37	SUBSCRIPTIONS = [
38	{"pubsubname": "taskflow-pubsub", "topic": "task-events", "route": "/dapr/events/task-events"},
39	{"pubsubname": "taskflow-pubsub", "topic": "reminders", "route": "/dapr/events/reminders"},
40	]
41
42	@router.get("/subscribe")
43	async def get_subscriptions() -> list[dict]:
44	"""Dapr calls this on startup to discover subscriptions."""
45	return SUBSCRIPTIONS
46	```
47
48	### 2. CloudEvent Handling (CRITICAL)
49
50	Dapr wraps all pub/sub messages in CloudEvent format. You MUST unwrap it.
51
52	```python
53	@router.post("/events/task-events")
54	async def handle_task_events(
55	request: Request,
56	session: AsyncSession = Depends(get_session),
57	) -> dict:
58	try:
59	# Step 1: Get raw CloudEvent
60	raw_event = await request.json()
61
62	# Step 2: ALWAYS unwrap CloudEvent "data" field
63	# CloudEvent structure:
64	# {
65	# "data": { <-- Your payload is HERE
66	# "event_type": "task.created",
67	# "data": {...},
68	# "timestamp": "..."
69	# },
70	# "datacontenttype": "application/json",
71	# "id": "...",
72	# "pubsubname": "taskflow-pubsub",
73	# "source": "myapp",
74	# "topic": "task-events",
75	# ...
76	# }
77	event = raw_event.get("data", raw_event) # Unwrap or use as-is
78
79	# Step 3: Now access your payload
80	event_type = event.get("event_type") # "task.created"
81	data = event.get("data", {}) # Your actual data
82
83	# Process event...
84
85	return {"status": "SUCCESS"}
86
87	except Exception as e:
88	logger.exception("Error handling event: %s", e)
89	# Return SUCCESS to prevent Dapr retries for bad events
90	return {"status": "SUCCESS"}
91	```
92
93	### 3. Publishing Events
94
95	```python
96	import httpx
97
98	DAPR_HTTP_ENDPOINT = "http://localhost:3500"
99	PUBSUB_NAME = "taskflow-pubsub"
100
101	async def publish_event(
102	topic: str,
103	event_type: str,
104	data: dict,
105	) -> bool:
106	"""Publish event to Dapr pub/sub."""
107	url = f"{DAPR_HTTP_ENDPOINT}/v1.0/publish/{PUBSUB_NAME}/{topic}"
108
109	payload = {
110	"event_type": event_type,
111	"data": data,
112	"timestamp": datetime.utcnow().isoformat(),
113	}
114
115	try:
116	async with httpx.AsyncClient(timeout=5.0) as client:
117	response = await client.post(url, json=payload)
118	return response.status_code == 204
119	except Exception as e:
120	logger.error("Failed to publish event: %s", e)
121	return False
122	```
123
124	### 4. Dapr Jobs API (Scheduled Jobs)
125
126	CRITICAL: Dapr Jobs v1.0-alpha1 calls back to `/job/{job_name}` by default!
127
128	```python
129	# Scheduling a job
130	async def schedule_job(
131	job_name: str,
132	due_time: datetime,
133	data: dict,
134	dapr_http_endpoint: str = "http://localhost:3500",
135	) -> bool:
136	"""Schedule a one-time Dapr job."""
137	url = f"{dapr_http_endpoint}/v1.0-alpha1/jobs/{job_name}"
138
139	payload = {
140	"dueTime": due_time.strftime("%Y-%m-%dT%H:%M:%SZ"), # RFC3339
141	"data": data,
142	}
143
144	try:
145	async with httpx.AsyncClient(timeout=5.0) as client:
146	response = await client.post(url, json=payload)
147	return response.status_code == 204
148	except Exception as e:
149	logger.error("Failed to schedule job: %s", e)
150	return False
151	```
152
153	Handling the callback - Dapr calls `/job/{job_name}`, NOT a custom endpoint:
154
155	```python
156	# WRONG - Dapr won't call this!
157	@router.post("/api/jobs/trigger")
158	async def handle_trigger(...):
159	pass
160
161	# CORRECT - This is what Dapr actually calls
162	@router.post("/job/{job_name}")
163	async def handle_dapr_job_callback(
164	job_name: str,
165	request: Request,
166	session: AsyncSession = Depends(get_session),
167	) -> dict:
168	"""Handle Dapr Jobs v1.0-alpha1 callback.
169
170	Da

People who liked this also viewed

6 picks

Type

dapr-integration

26stars11forks

View on GitHub

How to install dapr-integration?

mjunaidca/mjs-agent-skills/dapr-integration

$npx -y skills add mjunaidca/mjs-agent-skills --skill dapr-integration

Installs into the current project.

Summary

Files · 1

View on GitHub

SKILL.md

1	# Dapr Integration
2
3	Integrate Dapr sidecar for pub/sub messaging, state management, and scheduled jobs in Kubernetes environments.
4
5	## When to Use
6
7	- Setting up Dapr pub/sub for event-driven microservices
8	- Scheduling jobs with Dapr Jobs API (v1.0-alpha1)
9	- Handling CloudEvent message formats
10	- Implementing subscription handlers in FastAPI
11	- Debugging Dapr integration issues
12
13	## Quick Start
14
15	```bash
16	# Install Dapr CLI
17	curl -fsSL https://raw.githubusercontent.com/dapr/cli/master/install/install.sh \| bash
18
19	# Initialize Dapr (local)
20	dapr init
21
22	# Run app with Dapr sidecar
23	dapr run --app-id myapp --app-port 8000 -- uvicorn main:app
24	```
25
26	## Core Patterns
27
28	### 1. Pub/Sub Subscription Handler (FastAPI)
29
30	```python
31	from fastapi import APIRouter, Request
32	from sqlmodel.ext.asyncio.session import AsyncSession
33
34	router = APIRouter(prefix="/dapr", tags=["Dapr"])
35
36	# Topics we subscribe to
37	SUBSCRIPTIONS = [
38	{"pubsubname": "taskflow-pubsub", "topic": "task-events", "route": "/dapr/events/task-events"},
39	{"pubsubname": "taskflow-pubsub", "topic": "reminders", "route": "/dapr/events/reminders"},
40	]
41
42	@router.get("/subscribe")
43	async def get_subscriptions() -> list[dict]:
44	"""Dapr calls this on startup to discover subscriptions."""
45	return SUBSCRIPTIONS
46	```
47
48	### 2. CloudEvent Handling (CRITICAL)
49
50	Dapr wraps all pub/sub messages in CloudEvent format. You MUST unwrap it.
51
52	```python
53	@router.post("/events/task-events")
54	async def handle_task_events(
55	request: Request,
56	session: AsyncSession = Depends(get_session),
57	) -> dict:
58	try:
59	# Step 1: Get raw CloudEvent
60	raw_event = await request.json()
61
62	# Step 2: ALWAYS unwrap CloudEvent "data" field
63	# CloudEvent structure:
64	# {
65	# "data": { <-- Your payload is HERE
66	# "event_type": "task.created",
67	# "data": {...},
68	# "timestamp": "..."
69	# },
70	# "datacontenttype": "application/json",
71	# "id": "...",
72	# "pubsubname": "taskflow-pubsub",
73	# "source": "myapp",
74	# "topic": "task-events",
75	# ...
76	# }
77	event = raw_event.get("data", raw_event) # Unwrap or use as-is
78
79	# Step 3: Now access your payload
80	event_type = event.get("event_type") # "task.created"
81	data = event.get("data", {}) # Your actual data
82
83	# Process event...
84
85	return {"status": "SUCCESS"}
86
87	except Exception as e:
88	logger.exception("Error handling event: %s", e)
89	# Return SUCCESS to prevent Dapr retries for bad events
90	return {"status": "SUCCESS"}
91	```
92
93	### 3. Publishing Events
94
95	```python
96	import httpx
97
98	DAPR_HTTP_ENDPOINT = "http://localhost:3500"
99	PUBSUB_NAME = "taskflow-pubsub"
100
101	async def publish_event(
102	topic: str,
103	event_type: str,
104	data: dict,
105	) -> bool:
106	"""Publish event to Dapr pub/sub."""
107	url = f"{DAPR_HTTP_ENDPOINT}/v1.0/publish/{PUBSUB_NAME}/{topic}"
108
109	payload = {
110	"event_type": event_type,
111	"data": data,
112	"timestamp": datetime.utcnow().isoformat(),
113	}
114
115	try:
116	async with httpx.AsyncClient(timeout=5.0) as client:
117	response = await client.post(url, json=payload)
118	return response.status_code == 204
119	except Exception as e:
120	logger.error("Failed to publish event: %s", e)
121	return False
122	```
123
124	### 4. Dapr Jobs API (Scheduled Jobs)
125
126	CRITICAL: Dapr Jobs v1.0-alpha1 calls back to `/job/{job_name}` by default!
127
128	```python
129	# Scheduling a job
130	async def schedule_job(
131	job_name: str,
132	due_time: datetime,
133	data: dict,
134	dapr_http_endpoint: str = "http://localhost:3500",
135	) -> bool:
136	"""Schedule a one-time Dapr job."""
137	url = f"{dapr_http_endpoint}/v1.0-alpha1/jobs/{job_name}"
138
139	payload = {
140	"dueTime": due_time.strftime("%Y-%m-%dT%H:%M:%SZ"), # RFC3339
141	"data": data,
142	}
143
144	try:
145	async with httpx.AsyncClient(timeout=5.0) as client:
146	response = await client.post(url, json=payload)
147	return response.status_code == 204
148	except Exception as e:
149	logger.error("Failed to schedule job: %s", e)
150	return False
151	```
152
153	Handling the callback - Dapr calls `/job/{job_name}`, NOT a custom endpoint:
154
155	```python
156	# WRONG - Dapr won't call this!
157	@router.post("/api/jobs/trigger")
158	async def handle_trigger(...):
159	pass
160
161	# CORRECT - This is what Dapr actually calls
162	@router.post("/job/{job_name}")
163	async def handle_dapr_job_callback(
164	job_name: str,
165	request: Request,
166	session: AsyncSession = Depends(get_session),
167	) -> dict:
168	"""Handle Dapr Jobs v1.0-alpha1 callback.
169
170	Da

People who liked this also viewed

6 picks

Type