fix: restore MCP compatibility and confine type-map reads

fixtures/sample.alembic.py

@@ -4,24 +4,26 @@
 Revises:
 Create Date: 2026-05-15 03:00:00.000000
 """
+
 import sqlalchemy as sa
 from alembic import op
 
 # revision identifiers, used by Alembic.
-revision = 'sample'
+revision = "sample"
 down_revision = None
 
 
 def upgrade() -> None:
-    op.create_table('users',
-        sa.Column('id', sa.Integer(), primary_key=True),
-        sa.Column('name', sa.String(100), nullable=False),
-        sa.Column('email', sa.String(255), nullable=False, unique=True),
-        sa.Column('role', sa.Enum('admin', 'editor', 'viewer'), nullable=False),
-        sa.Column('is_active', sa.Boolean(), server_default=True),
-        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
+    op.create_table(
+        "users",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("name", sa.String(100), nullable=False),
+        sa.Column("email", sa.String(255), nullable=False, unique=True),
+        sa.Column("role", sa.Enum("admin", "editor", "viewer"), nullable=False),
+        sa.Column("is_active", sa.Boolean(), server_default=True),
+        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now()),
     )
 
 
 def downgrade() -> None:
-    op.drop_table('users')
+    op.drop_table("users")

fixtures/sample.django.py

@@ -7,10 +7,10 @@
 
 
 class UserRole(models.TextChoices):
-    ADMIN = 'admin', 'Admin'
-    EDITOR = 'editor', 'Editor'
-    AUTHOR = 'author', 'Author'
-    SUBSCRIBER = 'subscriber', 'Subscriber'
+    ADMIN = "admin", "Admin"
+    EDITOR = "editor", "Editor"
+    AUTHOR = "author", "Author"
+    SUBSCRIBER = "subscriber", "Subscriber"
 
 
 class User(models.Model):
@@ -26,7 +26,7 @@ class User(models.Model):
     created_at = models.DateTimeField(auto_now_add=True)
 
     class Meta:
-        db_table = 'users'
+        db_table = "users"
 
     def __str__(self):
         return self.name
@@ -38,8 +38,8 @@ class Category(models.Model):
     sort_order = models.IntegerField(default=0)
 
     class Meta:
-        db_table = 'categories'
-        verbose_name_plural = 'categories'
+        db_table = "categories"
+        verbose_name_plural = "categories"
 
     def __str__(self):
         return self.name
@@ -50,27 +50,27 @@ class Post(models.Model):
     slug = models.SlugField(max_length=255, unique=True)
     content = models.TextField()
     excerpt = models.TextField(null=True, blank=True, max_length=500)
-    status = models.CharField(max_length=20, default='draft')
+    status = models.CharField(max_length=20, default="draft")
     published_at = models.DateTimeField(null=True, blank=True)
     author = models.ForeignKey(
         User,
         on_delete=models.CASCADE,
-        related_name='posts',
+        related_name="posts",
     )
     category = models.ForeignKey(
         Category,
         on_delete=models.SET_NULL,
         null=True,
         blank=True,
-        related_name='posts',
+        related_name="posts",
     )
     view_count = models.IntegerField(default=0)
     rating = models.DecimalField(max_digits=3, decimal_places=2, default=0.00)
     created_at = models.DateTimeField(auto_now_add=True)
     updated_at = models.DateTimeField(auto_now=True)
 
     class Meta:
-        db_table = 'posts'
+        db_table = "posts"
 
     def __str__(self):
         return self.title

pyproject.toml

@@ -37,10 +37,10 @@ dependencies = [
 dev = [
     "pytest>=7.0",
     "pytest-cov>=4.0",
-    "mcp>=1.0",
+    "mcp>=1.0,<2",
     "ruff>=0.4.0",
 ]
-mcp = ["mcp>=1.0"]
+mcp = ["mcp>=1.0,<2"]
 
 [project.scripts]
 schemaforge = "schemaforge.cli:main"

scripts/check_consistency.py

@@ -1,4 +1,5 @@
 """Schema consistency check script for CI/CD pipelines."""
+
 import os
 import sys
 

src/schemaforge/check.py

@@ -73,9 +73,7 @@ def check_directory(
             failures.append(f"  FAIL {name}: {e}")
 
     if not converted:
-        return "FAIL: No files could be converted to {}\n{}".format(
-            canonical, "\n".join(failures)
-        )
+        return "FAIL: No files could be converted to {}\n{}".format(canonical, "\n".join(failures))
 
     # Compare pairwise
     mismatches: list[str] = []

src/schemaforge/diff.py

@@ -75,13 +75,7 @@ def diff_schemas(text_a: str, text_b: str, fmt: str) -> str:
         )
         for name in common
     )
-    if (
-        not added
-        and not removed
-        and not has_table_diffs
-        and not enum_added
-        and not enum_removed
-    ):
+    if not added and not removed and not has_table_diffs and not enum_added and not enum_removed:
         lines.append("No differences found.")
 
     return "\n".join(lines)

src/schemaforge/generators/alembic_generator.py

@@ -73,9 +73,7 @@ def generate(
             lines.append("# ### enum type definitions ###")
             for enum_type in schema.enums:
                 values = ", ".join(f"'{v}'" for v in enum_type.values)
-                lines.append(
-                    f'op.execute("CREATE TYPE {enum_type.name} AS ENUM ({values})")'
-                )
+                lines.append(f'op.execute("CREATE TYPE {enum_type.name} AS ENUM ({values})")')
             lines.append("")
 
         lines.append("# revision identifiers, used by Alembic.")
@@ -107,9 +105,7 @@ def generate(
             for table in reversed(schema.tables):
                 for idx in reversed(table.indexes):
                     idx_name = idx.name or f"idx_{'_'.join(idx.columns)}"
-                    lines.append(
-                        f"    op.drop_index('{idx_name}', table_name='{table.name}')"
-                    )
+                    lines.append(f"    op.drop_index('{idx_name}', table_name='{table.name}')")
             for table in reversed(schema.tables):
                 lines.append(f"    op.drop_table('{table.name}')")
             for enum_type in reversed(schema.enums):
@@ -139,14 +135,9 @@ def _generate_table(self, table: Table) -> tuple[list[str], list[str]]:
             col_list = ", ".join(f"'{c}'" for c in idx.columns)
             idx_name = idx.name or f"idx_{'_'.join(idx.columns)}"
             if idx.unique:
-                index_lines.append(
-                    f"    op.create_unique_constraint('{idx_name}', "
-                    f"'{table.name}', [{col_list}])"
-                )
+                index_lines.append(f"    op.create_unique_constraint('{idx_name}', '{table.name}', [{col_list}])")
             else:
-                index_lines.append(
-                    f"    op.create_index('{idx_name}', '{table.name}', [{col_list}])"
-                )
+                index_lines.append(f"    op.create_index('{idx_name}', '{table.name}', [{col_list}])")
 
         return table_lines, index_lines
 
@@ -176,16 +167,12 @@ def _column_def(self, col: Column) -> str:
             kwargs.append("unique=True")
 
         # fn: defaults (server_default)
-        fn_default = resolve_fn_default(
-            col, fn_wrapper="sa.func.{}", expr_fallback="sa.text('{}')"
-        )
+        fn_default = resolve_fn_default(col, fn_wrapper="sa.func.{}", expr_fallback="sa.text('{}')")
         if fn_default:
             kwargs.append(f"server_default={fn_default}")
 
         # Literal defaults
-        if col.default is not None and not (
-            isinstance(col.default, str) and col.default.startswith("fn:")
-        ):
+        if col.default is not None and not (isinstance(col.default, str) and col.default.startswith("fn:")):
             lit = format_literal_default(col)
             kwargs.append(f"server_default={lit}")
 

src/schemaforge/generators/django_generator.py

@@ -69,10 +69,7 @@ def _generate_model(self, table) -> str:
                     # Unique on single column is already handled via field option
                     pass
                 else:
-                    lines.append(
-                        f"    {col} = models.{'IntegerField' if col else 'CharField'}("
-                        f"db_index=True)"
-                    )
+                    lines.append(f"    {col} = models.{'IntegerField' if col else 'CharField'}(db_index=True)")
 
         # Unique constraints (multi-column)
         unique_sets: list[list[str]] = []
@@ -103,9 +100,7 @@ def _generate_model(self, table) -> str:
 
     def _field_def(self, col: Column) -> str:
         """Generate a Django model field definition."""
-        django_field = resolve_type(
-            col, self._FIELD_MAP, fmt="django", type_config=self._type_config
-        )
+        django_field = resolve_type(col, self._FIELD_MAP, fmt="django", type_config=self._type_config)
         if not django_field.endswith("Field"):
             django_field = django_field + "Field"
 

src/schemaforge/generators/drizzle_generator.py

@@ -27,9 +27,7 @@
 class DrizzleGenerator:
     """Generate Drizzle ORM TypeScript schema from Schema IR."""
 
-    def __init__(
-        self, dialect: str = "pg", type_config: TypeConfig | None = None
-    ) -> None:
+    def __init__(self, dialect: str = "pg", type_config: TypeConfig | None = None) -> None:
         """Initialize with dialect and optional custom type overrides.
 
         Args:
@@ -81,11 +79,7 @@ def _collect_imports(self, schema: Schema) -> str:
                 type_imports.add(drizzle_type)
 
                 # Add serial for primary keys with integer type
-                if (
-                    col.primary_key
-                    and col.type == ColumnType.INTEGER
-                    and not col.default
-                ):
+                if col.primary_key and col.type == ColumnType.INTEGER and not col.default:
                     type_imports.add("serial")
                     type_imports.discard("integer")
 

src/schemaforge/generators/json_schema_generator.py

@@ -163,9 +163,7 @@ def _column_to_prop(self, col: Column) -> dict[str, Any]:
 
         return self._add_base_annotations(prop, col)
 
-    def _add_base_annotations(
-        self, prop: dict[str, Any], col: Column
-    ) -> dict[str, Any]:
+    def _add_base_annotations(self, prop: dict[str, Any], col: Column) -> dict[str, Any]:
         """Add common annotations (description, default, comment) to a property."""
         if col.comment:
             prop["description"] = col.comment

src/schemaforge/generators/prisma_generator.py

@@ -34,10 +34,7 @@ def generate(self, schema: Schema) -> str:
         """Generate Prisma schema from IR."""
         parts: list[str] = []
         parts.append('generator client {\n  provider = "prisma-client-js"\n}')
-        parts.append(
-            'datasource db {\n  provider = "postgresql"\n'
-            '  url      = env("DATABASE_URL")\n}'
-        )
+        parts.append('datasource db {\n  provider = "postgresql"\n  url      = env("DATABASE_URL")\n}')
 
         # Generate enums
         for enum_type in schema.enums:
@@ -65,9 +62,7 @@ def _generate_model(self, table) -> str:
 
     def _field_def(self, col: Column) -> str:
         """Generate a Prisma field definition."""
-        prisma_type = resolve_type(
-            col, self._TYPE_MAP, fmt="prisma", type_config=self._type_config
-        )
+        prisma_type = resolve_type(col, self._TYPE_MAP, fmt="prisma", type_config=self._type_config)
 
         # Handle String with length (Prisma uses @db.VarChar) — skip if overridden
         if (

src/schemaforge/generators/sqlalchemy_generator.py

@@ -116,10 +116,7 @@ def _column_def(self, col: Column) -> tuple[list[str], set[str]]:
         # Primary key
         if col.primary_key:
             kwargs.append("primary_key=True")
-            if (
-                col.type != ColumnType.INTEGER
-                or col.type_args.get("autoincrement") is False
-            ):
+            if col.type != ColumnType.INTEGER or col.type_args.get("autoincrement") is False:
                 kwargs.append("autoincrement=False")
 
         # Nullable — SQLAlchemy defaults to True, so only emit when False
@@ -135,16 +132,12 @@ def _column_def(self, col: Column) -> tuple[list[str], set[str]]:
             kwargs.append("index=True")
 
         # fn: defaults (server_default)
-        fn_default = resolve_fn_default(
-            col, fn_wrapper="func.{}", expr_fallback="text('{}')"
-        )
+        fn_default = resolve_fn_default(col, fn_wrapper="func.{}", expr_fallback="text('{}')")
         if fn_default:
             kwargs.append(f"server_default={fn_default}")
 
         # Literal defaults
-        if col.default is not None and not (
-            isinstance(col.default, str) and col.default.startswith("fn:")
-        ):
+        if col.default is not None and not (isinstance(col.default, str) and col.default.startswith("fn:")):
             lit = format_literal_default(col)
             if isinstance(col.default, bool | int | float | str):
                 kwargs.append(f"default={lit}")

src/schemaforge/generators/typeorm_generator.py

@@ -61,21 +61,14 @@ def generate(self, schema: Schema) -> str:
         has_index = any(t.indexes for t in schema.tables)
         if has_index:
             imports.append("Index")
-        has_unique = any(
-            any(c.unique and not c.primary_key for c in t.columns)
-            for t in schema.tables
-        )
+        has_unique = any(any(c.unique and not c.primary_key for c in t.columns) for t in schema.tables)
         if has_unique:
             imports.append("Unique")
 
         if len(imports) == 1 and imports[0] == "Entity":
-            parts[0] = (
-                'import { Entity, PrimaryGeneratedColumn, Column } from "typeorm";'
-            )
+            parts[0] = 'import { Entity, PrimaryGeneratedColumn, Column } from "typeorm";'
         else:
-            parts[0] = (
-                "import { " + ", ".join(sorted(set(imports))) + ' } from "typeorm";'
-            )
+            parts[0] = "import { " + ", ".join(sorted(set(imports))) + ' } from "typeorm";'
 
         entities: list[str] = []
         for table in schema.tables:
@@ -121,9 +114,7 @@ def _column_def(self, col: Column) -> list[str]:
         options: dict[str, str] = {}
 
         # Determine TypeORM type
-        col_type = resolve_type(
-            col, self._TYPE_MAP, fmt="typeorm", type_config=self._type_config
-        )
+        col_type = resolve_type(col, self._TYPE_MAP, fmt="typeorm", type_config=self._type_config)
 
         # Primary key handling
         if col.primary_key: