Content
# Aegis DataFlow (Integrated Data Engineering Agent)
**Turn natural-language goals into validated, monitored, human-approved pipeline DAGs.**
Aegis DataFlow is an autonomous data-engineering AI agent that integrates 9 Open\* tools ([OpenIngest](https://github.com/shubhmartin107-web/openingest), [OpenCDC](https://github.com/shubhmartin107-web/opencdc), [OpenLake](https://github.com/shubhmartin107-web/openlake), [OpenStream](https://github.com/shubhmartin107-web/openstream), [OpenPipe](https://github.com/shubhmartin107-web/openpipe), [OpenQuality](https://github.com/shubhmartin107-web/openquality), [OpenCatalog](https://github.com/shubhmartin107-web/opencatalog), [OpenVectorDB](https://github.com/shubhmartin107-web/openvectordb), [OpenFeatureStore](https://github.com/shubhmartin107-web/openfeaturestore)) via the [Model Context Protocol (MCP)](https://modelcontextprotocol.io) into one unified orchestration system.
## Features
- **🤖 AI-Powered Pipeline Generation** — Describe your goal in natural language; the LLM planner generates a multi-step pipeline DAG
- **⚡ DAG Orchestration** — Parallel wave execution with topological sorting, configurable retry with exponential backoff, and per-stage timeouts
- **👁️ Human-in-the-Loop** — Approval gates pause execution at critical stages for manual sign-off
- **🔌 MCP Tool Bus** — Unified interface for all tools via stdio, HTTP, and custom protocol adapters
- **⏰ Scheduled Execution** — Cron-based trigger system for recurring pipelines
- **🔐 API Authentication** — Key-based auth with SHA-256 hashed secrets
- **📜 Pipeline Versioning** — Full version history with restore capability
- **🔗 Webhook Notifications** — HMAC-signed webhooks on pipeline completion/failure
- **🔁 Idempotent Runs** — Idempotency keys prevent duplicate pipeline executions
- **🔄 Concurrency Locking** — Automatic prevention of duplicate pipeline runs
- **📦 Import/Export** — JSON-based pipeline portability
- **📊 Real-Time SSE Events** — Live stream execution events to Web UI
- **🛡️ OpenTelemetry** — Distributed tracing and metrics
## Architecture
```
┌─────────────────────────────────────────────────────────┐
│ Aegis DataFlow │
├─────────────────────────────────────────────────────────┤
│ User Input (Goal / API / Schedule) │
│ │ │
│ ┌──────▼──────┐ ┌──────────────────┐ │
│ │ LLM Planner │─▶│ DAG Executor │ │
│ │ (Python) │ │ (Go) │ │
│ └──────────────┘ └───────┬──────────┘ │
│ │ │
│ ┌─────────────────────────▼──────────────────────────┐ │
│ │ MCP Tool Bus │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ stdio │ │ HTTP │ │ Custom │ │ │
│ │ │ adapter │ │ adapter │ │ adapter │ │ │
│ │ └────┬────┘ └────┬────┘ └────┬────┘ │ │
│ └───────┼───────────┼───────────┼──────────────────────┘ │
│ │ │ │ │
│ ┌───────▼───┐ ┌────▼────┐ ┌───▼────────┐ │
│ │ OpenIngest│ │ OpenLake│ │ OpenCDC │ ... 9 tools │
│ └───────────┘ └─────────┘ └────────────┘ │
│ │
│ ┌────────────────────────────────────────────────────┐ │
│ │ State Store (SQLite) │ │
│ │ Pipelines │ Versions │ Runs │ Approval Gates │ │
│ │ Webhooks │ Schedules │ API Keys │ Tool Registry │ │
│ └────────────────────────────────────────────────────┘ │
│ │
│ ┌────────────────────────────────────────────────────┐ │
│ │ Web UI (React + TypeScript) │ │
│ │ Dashboard │ Pipeline Graph │ Goal Input │ Logs │ │
│ └────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘
```
## Quick Start
### Prerequisites
- Go 1.24+ ([install](https://go.dev/dl/))
- Python 3.10+ for the LLM planner
- [Ollama](https://ollama.ai) with a model (default: `qwen2.5:7b`)
- Node.js 20+ for the Web UI (optional)
### 1. Clone & Build
```bash
git clone https://github.com/shubhmartin107-web/aegis-dataflow.git
cd aegis-dataflow
make build
```
### 2. Configure
```bash
cp config.example.json config.json
# Edit config.json to point to your MCP tool servers
# Or set environment variables:
export AEGIS_OLLAMA_ENDPOINT=http://localhost:11434
export AEGIS_PLANNER_MODEL=qwen2.5:7b
```
### 3. Run
```bash
# Start the server
./bin/aegis
# With config:
./bin/aegis -config config.json
```
The server starts at `http://0.0.0.0:9876`.
## Configuration
| Environment Variable | Default | Description |
|---|---|---|
| `AEGIS_HOST` | `0.0.0.0` | Server bind address |
| `AEGIS_PORT` | `9876` | Server port |
| `AEGIS_OLLAMA_ENDPOINT` | `http://localhost:11434` | Ollama API endpoint |
| `AEGIS_PLANNER_MODEL` | `qwen2.5:7b` | LLM model for pipeline planning |
| `AEGIS_AUTH_ENABLED` | `false` | Enable API key authentication |
| `AEGIS_AI_API_KEY` | — | API key for external AI providers |
## API Reference
### Pipelines
| Method | Path | Description |
|---|---|---|
| `GET` | `/api/pipelines` | List pipelines (`?status=draft\|active`) |
| `POST` | `/api/pipelines` | Create pipeline from JSON |
| `GET` | `/api/pipelines/{id}` | Get pipeline details |
| `PUT` | `/api/pipelines/{id}` | Update pipeline |
| `DELETE` | `/api/pipelines/{id}` | Delete pipeline |
| `POST` | `/api/pipelines/{id}/run` | Execute pipeline (supports `idempotency_key`) |
| `GET` | `/api/pipelines/{id}/runs` | List pipeline runs |
| `GET` | `/api/pipelines/{id}/versions` | List version history |
| `POST` | `/api/pipelines/{id}/restore/{v}` | Restore a previous version |
| `GET` | `/api/pipelines/{id}/export` | Export pipeline as JSON |
### Goals
| Method | Path | Description |
|---|---|---|
| `POST` | `/api/goals` | Submit natural-language goal for planning |
### Approvals
| Method | Path | Description |
|---|---|---|
| `GET` | `/api/approvals/{runID}` | Get pending approval gates |
| `POST` | `/api/approvals/{gateID}/approve` | Approve a stage |
| `POST` | `/api/approvals/{gateID}/reject` | Reject a stage |
### Webhooks & Schedules
| Method | Path | Description |
|---|---|---|
| `POST/GET/DELETE` | `/api/pipelines/{id}/webhooks` | Webhook CRUD |
| `POST/GET/DELETE` | `/api/pipelines/{id}/schedules` | Cron schedule CRUD |
### Admin
| Method | Path | Description |
|---|---|---|
| `POST/GET/DELETE` | `/api/api-keys` | API key management |
| `POST` | `/api/import` | Import pipeline from JSON |
| `GET` | `/api/events` | SSE stream of execution events |
### MCP Interface
| Method | Path | Description |
|---|---|---|
| `GET` | `/mcp` | MCP SSE endpoint |
| `POST` | `/mcp` | MCP JSON-RPC endpoint |
## Development
```bash
# Build
make build
# Test
make test
# Run all checks
make lint
# Build Web UI
make ui-build
```
## Technology Stack
| Layer | Language | Libraries |
|---|---|---|
| Orchestration | Go | chi, go-sqlite3, dominikbraun/graph |
| AI Planner | Python | ollama, pydantic |
| Web UI | TypeScript/React | React Flow, Vite |
| Storage | SQLite | WAL mode, foreign keys |
| Observability | Go | OpenTelemetry |
| Protocol | — | MCP 2025-11-25, JSON-RPC 2.0 |
## License
Apache 2.0 — see [LICENSE](LICENSE).
## Project Status
v0.1.0 — Core specification complete. All 9 Open\* tools supported via MCP adapters. Ready for local deployment.
