contd 0.1.3

Durable workflow execution engine with time-travel debugging

Contd.ai

The Durable Execution Engine for Agentic Workflows.

Contd.ai is a lightweight, multi-tenant framework designed to build, run, and recover long-running AI agent workflows. It provides durable execution guarantees, meaning your agents can pause, resume, crash, and recover exactly where they left off—without losing state or context.

🚀 Key Features

• Durable Execution: Workflows are resumable by default. State is persisted after every step.
• Multi-Tenancy: Built-in organization isolation. Data and execution contexts are strictly scoped to organizations.
• Epistemic Savepoints: specialized markers for AI agents to save their logic state (hypotheses, goals, decisions) alongside execution state.
• Hybrid Recovery: Fast restoration using snapshots + event replay.
• Idempotency: Automatic retries and de-duplication of steps.
• Secure API: JWT-based user authentication and Scoped API Keys for service interactions.

🛠️ Architecture

Contd.ai follows a layered architecture:

• SDK (contd.sdk): Python decorators (@workflow, @step) to define workflows. Handles context propagation and serialization.
• Engine (contd.core): The brain. Manages the event loop, recovery logic, and lease management.
• API (contd.api): FastAPI interactions for submitting workflows, querying status, and time-travel.
• Persistence (contd.persistence): Pluggable storage adapters. Currently supports Postgres (Journal/Leases) and S3 (Snapshots).

📦 Project Structure

contd/
├── api/            # FastAPI routes & gRPC extensions
├── core/           # Execution Engine, Auth, Recovery
├── models/         # Pydantic & Dataclass entity definitions
├── persistence/    # DB Adapters (SQL, S3) & Journal logic
├── sdk/            # User-facing decorators & Context
└── observability/  # Metrics & Tracing

sdks/
├── typescript/     # TypeScript/Node.js SDK
├── go/             # Go SDK
└── java/           # Java SDK (Enterprise)

📚 Client Libraries

Contd provides SDKs for multiple languages:

Language	Package	Status
Python	contd.sdk	✅ Primary (built-in)
TypeScript/Node.js	@contd/sdk	✅ Available
Go	github.com/contd/sdk-go	✅ Available
Java	ai.contd:contd-sdk	✅ Available

TypeScript/Node.js

npm install @contd/sdk

import { ContdClient, workflow, step } from '@contd/sdk';

const client = new ContdClient({ apiKey: 'your-api-key' });
const workflowId = await client.startWorkflow({
  workflowName: 'process-order',
  input: { orderId: '12345' }
});

Go

go get github.com/contd/sdk-go

import contd "github.com/contd/sdk-go"

client := contd.NewClient(contd.ClientConfig{APIKey: "your-api-key"})
workflowID, _ := client.StartWorkflow(ctx, contd.StartWorkflowInput{
    WorkflowName: "process-order",
    Input: map[string]interface{}{"orderId": "12345"},
})

Java

<dependency>
    <groupId>ai.contd</groupId>
    <artifactId>contd-sdk</artifactId>
    <version>1.0.0</version>
</dependency>

ContdClient client = ContdClient.builder()
    .apiKey("your-api-key")
    .build();
String workflowId = client.startWorkflow("process-order", 
    Map.of("orderId", "12345"), null);

See the sdks/ directory for full documentation on each SDK.

🏁 Getting Started

Prerequisites

• Python 3.10+
• PostgreSQL (Local or Remote)

Installation

# Clone the repository
git clone https://github.com/bhavd98/contd.ai.git
cd contd.ai

# Install dependencies
pip install -r requirements.txt

Database Setup

1. Create a PostgreSQL database.
2. Run the schema script to initialize tables:

psql -d contd_db -f contd/persistence/schema.sql

Running the Server

Start the API and gRPC server:

python -m contd.api.server

The server listens on:

• HTTP API: http://localhost:8000
• gRPC: localhost:50051

🔐 Authentication

1. User Signup & Login

Contd uses JWT for user sessions.

Signup:

POST /v1/auth/signup
Content-Type: application/json

{
  "email": "[e-mail]",
  "password": "securepassword",
  "full_name": "Developer"
}

Login:

POST /v1/auth/token
Content-Type: application/x-www-form-urlencoded

username=[e-mail]&password=securepassword

Returns: access_token

2. Create Organization & API Key

Create an organization to scope your workflows.

Create Org:

POST /v1/auth/organizations
Authorization: Bearer <access_token>

{ "name": "My AI Team" }

Generate API Key:

POST /v1/auth/apikeys
Authorization: Bearer <access_token>
X-Organization-Id: <org_id>

{ "name": "Production Agent", "scopes": ["workflow:write"] }

👩‍💻 Usage Example

Defining a Workflow

from contd.sdk.decorators import workflow, step
from contd.sdk.context import ExecutionContext

@step
def research_topic(topic: str):
    # Simulate work
    return {"summary": f"Research on {topic} complete."}

@step
def draft_content(context: dict):
    # Uses previous step output
    return {"draft": f"Draft based on {context['summary']}"}

@workflow
def blog_agent(topic: str):
    data = research_topic(topic)
    final = draft_content(data)
    return final

Triggering a Workflow

Use your API Key to trigger the workflow via HTTP:

POST /v1/workflows
X-API-Key: sk_live_...

{
  "workflow_name": "blog_agent",
  "input": { "topic": "AI Agents" }
}

💻 CLI Usage

Contd includes a powerful CLI for local development and workflow management.

Installation

pip install -e .

Initialize a Project

contd init                    # Initialize with SQLite (default)
contd init --backend postgres # Initialize with Postgres

Run a Workflow

contd run my_workflow --input '{"key": "value"}'
contd run my_workflow -f input.json

Check Workflow Status

contd status <workflow_id>

Resume a Suspended Workflow

contd resume <workflow_id>

Inspect Workflow State

contd inspect <workflow_id>           # Basic info
contd inspect <workflow_id> --verbose # Full state and events

Time-Travel Debugging

contd time-travel <workflow_id> <savepoint_id>           # Create new workflow from savepoint
contd time-travel <workflow_id> <savepoint_id> --dry-run # Preview without creating

View Execution Logs

contd logs <workflow_id>
contd logs <workflow_id> -n 100 -l DEBUG  # More lines, debug level

List Workflows

contd list                    # All workflows
contd list --status running   # Filter by status

🧪 Testing

Run the test suite:

pytest tests/

📖 Documentation

Document	Description
Quickstart Guide	Get started in 5 minutes
Architecture	System design and components
API Reference	Complete REST/gRPC/SDK reference
Metrics Setup	Observability configuration
Troubleshooting	Common issues and solutions
Contributing	How to contribute

Migration Guides

From	Guide
LangChain	Migration Guide
Temporal	Migration Guide

Examples

See the examples/ directory for 12+ real-world workflow examples:

• Basic pipelines and error handling
• AI agents with tools
• RAG pipelines
• E-commerce order processing (saga pattern)
• ETL workflows
• Human-in-the-loop approvals
• Batch processing
• Webhook integrations
• Research and code review agents
• Customer support automation