Governed scheduling for Python agents — CrewAI, LangGraph, and custom agents in 2 lines.
Project description
dingdawg-loop
Governed scheduling for Python agents in 2 lines. Works with CrewAI, LangGraph, AutoGen, and any custom agent framework.
from dingdawg_loop import schedule_governed
schedule_governed(agent_id="compliance-scanner", cron="0 9 * * *", verifier="neurosymbolic")
Every execution runs through the DingDawg governance gate before anything fires. Deny = nothing executes. No exceptions.
Prerequisites
Node.js >= 18 and the DingDawg Loop MCP server:
npm install -g dingdawg-loop
Then install this package:
pip install dingdawg-loop
Use cases
1 — Daily compliance report (CrewAI)
from dingdawg_loop import schedule_governed, execute_governed
from crewai import Crew, Agent, Task
compliance_crew = Crew(agents=[...], tasks=[...])
def run_compliance(loop_id: str):
compliance_crew.kickoff()
loop = schedule_governed(
agent_id="compliance-scanner",
cron="0 9 * * *", # 09:00 every day
action_type="generate_report",
risk_tier="medium",
description="Daily SOC-2 compliance scan and PDF report generation",
verifier="neurosymbolic",
on_execute=run_compliance,
)
print(loop)
# {"loop_id": "abc123", "status": "registered", "schedule": "0 9 * * *"}
2 — Recurring data sync (LangGraph)
from dingdawg_loop import schedule_governed, execute_governed
loop = schedule_governed(
agent_id="data-sync-agent",
cron="*/15 * * * *", # every 15 minutes
action_type="data_sync",
risk_tier="low",
description="Sync CRM records to analytics warehouse",
verifier="neurosymbolic",
)
# Manually trigger when needed (governance gate runs first)
result = execute_governed(loop["loop_id"], notes="Manual sync before board meeting")
print(result["outcome"]) # "executed" | "denied" | "review_queued" | "error_skipped"
3 — High-risk financial action with inline gate
from dingdawg_loop import GovernanceGate
gate = GovernanceGate(fail_closed=True)
receipt = gate.govern(
agent_id="finance-agent",
action_type="wire_transfer",
description="Transfer $42,000 to vendor account #8821",
risk_tier="critical",
)
if receipt["decision"] == "allow":
execute_wire_transfer(...)
else:
notify_ops_team(receipt)
API reference
schedule_governed(...) -> dict
Register a governed loop. Returns {"loop_id", "status", "schedule"}.
| Param | Type | Default | Description |
|---|---|---|---|
agent_id |
str | required | Unique agent identifier |
cron |
str | required | Standard cron expression |
action_type |
str | "scheduled_execution" |
What the agent does |
risk_tier |
str | "low" |
"low" / "medium" / "high" / "critical" |
description |
str | "" |
Human-readable description |
verifier |
str | "neurosymbolic" |
"neurosymbolic" or "local" |
on_execute |
callable | None |
Callback fired on governance allow |
fail_closed |
bool | True |
Skip execution if governance unreachable |
execute_governed(loop_id, notes="") -> dict
Manually trigger a governed loop. Governance runs first.
Returns {"outcome", "receipt_id", "risk_score"}.
Possible outcomes: "executed" / "denied" / "review_queued" / "error_skipped"
list_loops(agent_id=None) -> list[dict]
List registered loops. Optionally filter by agent_id.
Degrades gracefully to in-process registry if MCP is unreachable.
GovernanceGate
Inline governance check for one-off actions outside of a scheduled loop.
gate = GovernanceGate(fail_closed=True, timeout=5.0)
receipt = gate.govern(agent_id, action_type, description, risk_tier)
# receipt["decision"] == "allow" | "deny" | "review"
Fail-closed behaviour
By default all functions are fail-closed: if the npx dingdawg-loop MCP
process is unreachable, times out, or returns an error, execution is
skipped and the function returns {"outcome": "error_skipped", "reason": "..."}.
To disable (not recommended for production):
schedule_governed(..., fail_closed=False)
GovernanceGate(fail_closed=False)
Environment variables
| Variable | Default | Description |
|---|---|---|
DINGDAWG_NPX_CMD |
npx |
Override the npx binary path |
DINGDAWG_LOOP_PKG |
dingdawg-loop |
Override the npm package name |
Requirements
- Python >= 3.8
- Node.js >= 18
npm install -g dingdawg-loop- Zero Python dependencies (stdlib only)
License
MIT — ISG / DingDawg
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file dingdawg_loop-1.0.0.tar.gz.
File metadata
- Download URL: dingdawg_loop-1.0.0.tar.gz
- Upload date:
- Size: 9.6 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.12.3
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
163bbbee5188ff7480ccb7d1aa9955a06258ed25304770b3a75acc37daf57745
|
|
| MD5 |
e553250c9a177639b430b41d120b1d84
|
|
| BLAKE2b-256 |
a4669c158fdb777131e98fa21bf313ee9827a7b032430b3543bd4698886c017f
|
File details
Details for the file dingdawg_loop-1.0.0-py3-none-any.whl.
File metadata
- Download URL: dingdawg_loop-1.0.0-py3-none-any.whl
- Upload date:
- Size: 9.3 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.12.3
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
99219913fb1ffe5db93b547e2adc797a76cf29abc89df53ea65fb0e4cb19fa58
|
|
| MD5 |
6b3550b5713fa923253d177c2e0419f7
|
|
| BLAKE2b-256 |
7efc4c804ec87c92a2d83408a1347297f0dd79acf4da1bcc1f757e50cc3a04b4
|