Schemas & Validation

Every MCP tool has a clear input/output contract. All tools return JSON strings.

Tool Schemas

read_dag

Read an Airflow DAG file and return raw source with metadata.

Input:

{
  "type": "object",
  "properties": {
    "path": {
      "type": "string",
      "description": "Path to the Airflow DAG file"
    },
    "content": {
      "type": "string",
      "description": "Inline DAG source code (alternative to path)"
    }
  }
}

Output:

{
  "type": "object",
  "properties": {
    "source": { "type": "string", "description": "Raw DAG source code" },
    "file_path": { "type": ["string", "null"], "description": "Resolved absolute path" },
    "file_size_bytes": { "type": "integer" },
    "line_count": { "type": "integer" }
  },
  "required": ["source", "line_count"]
}

Error:

{
  "error": "File not found: dags/missing.py",
  "source": null
}

lookup_concept

Query pre-compiled Airflow→Prefect translation knowledge.

Input:

{
  "type": "object",
  "properties": {
    "concept": {
      "type": "string",
      "description": "Airflow concept to look up (e.g. 'PythonOperator', 'XCom', 'postgres')"
    }
  },
  "required": ["concept"]
}

Output:

{
  "type": "object",
  "properties": {
    "concept_type": {
      "type": "string",
      "enum": ["operator", "pattern", "connection", "concept"]
    },
    "airflow": { "type": "string" },
    "prefect_equivalent": { "type": "string" },
    "translation_rules": { "type": "array", "items": { "type": "string" } },
    "source": { "type": "string", "enum": ["colin", "fallback"] }
  }
}

search_prefect_docs

Search current Prefect documentation via the Prefect MCP server.

Input:

{
  "type": "object",
  "properties": {
    "query": {
      "type": "string",
      "description": "Search query for Prefect docs"
    }
  },
  "required": ["query"]
}

Output (success):

{
  "type": "object",
  "properties": {
    "results": { "type": "array", "items": { "type": "string" } },
    "query": { "type": "string" },
    "source": { "type": "string" }
  }
}

Output (error):

{
  "error": "Cannot connect to Prefect MCP at https://docs.prefect.io/mcp. Run 'colin run' for cached context."
}

validate

Syntax-check generated code and return both sources for comparison.

Input:

{
  "type": "object",
  "properties": {
    "original_dag": {
      "type": "string",
      "description": "Path or inline content of the original Airflow DAG"
    },
    "converted_flow": {
      "type": "string",
      "description": "Path or inline content of the generated Prefect flow"
    }
  },
  "required": ["original_dag", "converted_flow"]
}

Output:

{
  "type": "object",
  "properties": {
    "original_source": { "type": "string" },
    "converted_source": { "type": "string" },
    "syntax_valid": { "type": "boolean" },
    "syntax_errors": {
      "type": ["array", "null"],
      "items": {
        "type": "object",
        "properties": {
          "line": { "type": "integer" },
          "column": { "type": "integer" },
          "message": { "type": "string" }
        }
      }
    },
    "comparison_guidance": { "type": "string" }
  },
  "required": ["original_source", "converted_source", "syntax_valid"]
}

scaffold

Generate a Prefect project directory structure.

Input:

{
  "type": "object",
  "properties": {
    "output_directory": { "type": "string", "description": "Where to create the project" },
    "project_name": { "type": ["string", "null"], "description": "Project name (defaults to directory name)" },
    "workspace": { "type": "string", "default": "default", "description": "Workspace name for deployments/<workspace>/ structure" },
    "flow_names": { "type": ["array", "null"], "items": { "type": "string" }, "description": "List of flow names to create directories for" },
    "include_docker": { "type": "boolean", "default": true },
    "include_github_actions": { "type": "boolean", "default": true },
    "schedule_interval": { "type": ["string", "null"], "description": "Cron string, preset (@daily etc.), seconds, or None" }
  },
  "required": ["output_directory"]
}

Output:

{
  "type": "object",
  "properties": {
    "created_directories": { "type": "array", "items": { "type": "string" } },
    "created_files": { "type": "array", "items": { "type": "string" } },
    "prefect_yaml_template": { "type": "string" },
    "next_steps": { "type": "array", "items": { "type": "string" } }
  }
}

generate_deployment

Write a prefect.yaml deployment configuration from DAG metadata.

Input:

{
  "type": "object",
  "properties": {
    "output_directory": { "type": "string", "description": "Directory to write prefect.yaml into" },
    "flows": {
      "type": "array",
      "items": { "type": "object" },
      "description": "List of flow dicts, each with flow_name and entrypoint"
    },
    "workspace": { "type": "string", "default": "default", "description": "Workspace name" }
  },
  "required": ["output_directory", "flows"]
}

Output:

{
  "type": "object",
  "properties": {
    "created_file": { "type": "string" },
    "deployment_names": { "type": "array", "items": { "type": "string" } },
    "next_steps": { "type": "array", "items": { "type": "string" } }
  }
}

generate_migration_report

Write MIGRATION.md — a human-readable record of a DAG conversion.

Input:

{
  "type": "object",
  "properties": {
    "output_directory": { "type": "string", "description": "Directory to write MIGRATION.md into" },
    "dag_path": { "type": "string", "description": "Path to the original Airflow DAG file" },
    "flow_path": { "type": "string", "description": "Path to the generated Prefect flow file" },
    "decisions": {
      "type": "array",
      "items": { "type": "object" },
      "description": "List of dicts with component, outcome, rationale, manual_action"
    },
    "manual_actions": {
      "type": ["array", "null"],
      "items": { "type": "string" },
      "description": "Top-level action types not tied to a specific component"
    }
  },
  "required": ["output_directory", "dag_path", "flow_path", "decisions"]
}

Output:

{
  "type": "object",
  "properties": {
    "created_file": { "type": "string" },
    "checklist_items_count": { "type": "integer" }
  }
}

Error Conventions

All tools use the same error pattern — a JSON object with an error key:

{
  "error": "Description of what went wrong"
}

The error field contains a human-readable message. Some tools include additional context (e.g., source: null in read_dag errors).