Docs
AI Docs
Crystal Agents
Crystal Agents - System Documentation

Crystal Agents - System Documentation

Architecture Overview

Crystal Agents is a modular workflow-driven automation system. The architecture follows a layered approach:

┌─────────────────────────────────────────────────────────────────┐
│                      Presentation Layer                         │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────────┐  │
│  │   Dashboard  │  │ Telegram Bot │  │       CLI              │  │
│  └─────────────┘  └─────────────┘  └─────────────────────────┘  │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                     Orchestration Layer                         │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────────┐  │
│  │   Router    │  │   Runner    │  │   Approval System      │  │
│  └─────────────┘  └─────────────┘  └─────────────────────────┘  │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                        Agent Layer                              │
│  ┌─────────────────────────────────────────────────────────────┐│
│  │  AgentRegistry                                              ││
│  │  CEO | Developer | Tester | Research | Content | Reviewer  ││
│  └─────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                        Tool Layer                                │
│  ┌─────────────────────────────────────────────────────────────┐│
│  │  ToolRegistry                                               ││
│  │  RepoReader | RepoWriter | TestRunner | WebSearch | GitOps ││
│  └─────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                      Workflow Layer                              │
│  ┌─────────────────────────────────────────────────────────────┐│
│  │  WorkflowRegistry                                           ││
│  │  ContentDrafting | FeatureSpec | BugTriage | etc.          ││
│  └─────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────┘

Core Components

1. Orchestrator (orchestrator/)

The orchestration layer manages workflow execution, routing, and agent coordination.

Key Files:

  • orchestrator.py - Main orchestrator class
  • router.py - Maps tasks to workflows
  • runner.py - Executes workflows
  • ceo_router.py - Natural language task routing
  • types.py - Core data types (TaskRequest, RunRecord)

Runtime Flow:

  1. TaskRequest is received via CLI, dashboard, or Telegram
  2. Router maps task to appropriate workflow
  3. Runner executes workflow steps using agents
  4. Agents request tools via ToolExecutor
  5. Approval gates intercept approval-required tools
  6. Results are stored in runs/ directory

2. Agents (agents/)

Specialized AI agents for different tasks. Each agent has specific allowed tools defined in config/policy.json.

Base Agent Structure:

class BaseAgent:
    name: str
    allowed_tools: tuple[str, ...]
    model_config: ModelConfig
    
    def execute(self, context, prompt) -> str:
        """Execute agent task"""

Agent Implementations:

  • ceo.py - CEO router, task classification
  • developer.py - Code development
  • tester.py - Test execution
  • research.py - Web research
  • content.py - Content creation
  • reviewer.py - Code/content review
  • devops.py - DevOps operations
  • ads_manager.py - Ad campaign management
  • sales_assistant.py - Sales support
  • seo.py - SEO tasks
  • analytics.py - Data analysis

3. Tools (tools/)

Tools provide capabilities to agents. Each tool has configurable approval requirements.

Available Tools:

  • repo_reader.py - Read files from repository (no approval needed)
  • repo_writer.py - Write files to repository (requires approval)
  • test_runner.py - Execute tests (no approval needed)
  • web_search.py - Search the web (no approval needed)
  • git_ops.py - Git operations: branch, commit, push, PR (requires approval)

Tool Execution Flow:

ToolExecutor.execute(agent, tool, context, **kwargs):
    1. Check agent has permission for tool
    2. If tool requires approval:
       - Log approval request
       - Call approval_resolver
       - If approved, add to approval_gate
       - If denied, raise ApprovalRequiredError
    3. Execute tool.run(context, **kwargs)

4. Workflows (workflows/)

Workflows define multi-step processes for specific tasks.

Base Workflow:

class BaseWorkflow:
    name: str
    description: str
    
    def run(self, task, runtime) -> RunRecord:
        """Execute workflow steps"""

Available Workflows:

  • content_drafting.py - Generate marketing content
  • feature_spec.py - Create technical specifications
  • content_review.py - Review existing content
  • lead_followup.py - Lead follow-up communications
  • campaign_creation.py - Marketing campaigns
  • bug_triage.py - Bug report analysis
  • release_preparation.py - Release documentation
  • customer_reporting.py - Customer reports
  • proposal_generation.py - Business proposals
  • git_workflow.py - Git operations workflow

5. Scheduler (scheduler/)

Cron-based job scheduler for automated tasks.

Components:

  • service.py - Main scheduler service
  • cron.py - Cron expression parsing
  • models.py - Data models
  • history.py - Execution history tracking

Features:

  • Cron-based scheduling
  • Script validation against allowed roots
  • Optional approval requirements
  • Execution logging to logs/scheduler_history.jsonl
  • Telegram notifications on completion/failure

6. Dashboard (dashboard/)

Web-based UI for managing workflows and approvals.

Features:

  • Recent runs display
  • Workflow triggering
  • Pending approvals management
  • Scheduler monitoring
  • Daily/weekly reports

7. Notifications (notifications/)

Telegram integration for notifications.

Features:

  • Workflow completion notifications
  • Scheduler job status updates
  • Approval request notifications

Configuration

config/config.json 

{
  "telegram": { "enabled": true, "bot_token": "", "chat_id": "" },
  "model": { 
    "provider": "google",
    "name": "gemini/gemini-2.5-flash-lite",
    "temperature": 0.1,
    "max_tokens": 4096
  },
  "scheduler": {
    "enabled": true,
    "poll_interval_seconds": 60,
    "allowed_script_roots": ["scripts"],
    "jobs": [...]
  },
  "git": {
    "workspace_root": "...",
    "allowed_repos": ["owner/repo"]
  },
  "agent_models": { "developer": { "provider": "google", "name": "..." } }
}

config/policy.json 

{
  "agents": {
    "developer": { "allowed_tools": ["repo_reader", "repo_writer", "test_runner", "git_ops"] }
  },
  "tools": {
    "repo_writer": { "requires_approval": true },
    "git_ops": { "requires_approval": true }
  }
}

Data Flow

Task Execution Flow 

User Input (CLI/Dashboard/Telegram)
         │
         ▼
TaskRequest { task_id, task_type, prompt, metadata }
         │
         ▼
Orchestrator.run(task)
         │
         ▼
WorkflowRunner.execute()
         │
         ├─► Step 1: Agent executes with context
         │         │
         │         ▼
         │    ToolExecutor.execute()
         │         │
         │         ├─► Tool validation
         │         ├─► Approval check (if required)
         │         └─► Tool.run()
         │
         ├─► Step 2: (repeat for each step)
         │
         ▼
RunRecord { run_id, status, outputs, metadata }
         │
         ▼
Save to runs/{run_id}/
         │
         ▼
Notification (if enabled)

Approval Flow 

Agent requests tool execution
         │
         ▼
ToolExecutor checks requires_approval
         │
         ▼
Log approval request to approvals.jsonl
         │
         ▼
approval_resolver(agent, tool, context, kwargs)
         │
         ├─► True: Approve action, execute tool
         │
         └─► False: Deny action, log denial, raise error

Data Models

TaskRequest 

{
    task_id: str,
    task_type: str,  # content, feature-spec, etc.
    prompt: str,
    metadata: dict  # source, title, model info
}

RunRecord 

{
    run_id: str,
    workflow_name: str,
    status: str,  # running, succeeded, failed
    task: TaskRequest,
    outputs: list[str],
    metadata: dict  # started_at, run_dir, etc.
}

ApprovalEvent 

{
    event_type: str,  # requested, approved, denied
    action_key: str,  # unique action identifier
    run_id: str,
    agent_name: str,
    tool_name: str,
    step_name: str,
    requested_at: str,
    decided_by: str,
    reason: str
}

Security Model

Tool Access Control

  • Agents have whitelisted tools in config/policy.json
  • ToolExecutor validates tool access before execution

Approval System

  • Tools requiring approval are flagged in policy
  • Approval requests logged to logs/approvals.jsonl
  • Dashboard provides UI for approve/deny decisions

Script Validation

  • Scheduler scripts must be under allowed_script_roots
  • Path traversal prevention via workspace root validation

Extension Points

Adding New Agents

  1. Create agent class extending BaseAgent
  2. Define tool permissions in config/policy.json
  3. Register in orchestrator/policy.py

Adding New Tools

  1. Create tool class extending BaseTool
  2. Configure approval requirements in policy
  3. Register in tool registry

Adding New Workflows

  1. Create workflow class extending BaseWorkflow
  2. Define steps and agent assignments
  3. Register in WorkflowRegistry

Dependencies

  • Python: 3.11+
  • Key Libraries:
    • google-generativeai - Google AI models
    • openai - OpenAI models
    • python-telegram-bot - Telegram integration
    • Standard library: json, subprocess, http.server, pathlib
Crystal Agentic Team - Implementation RoadmapCrystal Agents - User Documentation