Add MCP tools for migration inspection (bd-627d Phase 3)

- Add inspect_migration() tool - calls bd migrate --inspect --json
- Add get_schema_info() tool - calls bd info --schema --json
- Implements abstract methods in BdClientBase
- CLI client calls commands directly
- Daemon client raises NotImplementedError (rare admin commands)

Phase 3 complete. Agents can now inspect migrations via MCP before running them.

Amp-Thread-ID: https://ampcode.com/threads/T-de7e1141-87ac-4b4a-9cea-1b7bc4d51da9
Co-authored-by: Amp <amp@ampcode.com>

integrations/beads-mcp/src/beads_mcp/bd_client.py

@@ -129,6 +129,16 @@ class BdClientBase(ABC):
         """Initialize a new beads database."""
         pass
 
+    @abstractmethod
+    async def inspect_migration(self) -> dict:
+        """Get migration plan and database state for agent analysis."""
+        pass
+
+    @abstractmethod
+    async def get_schema_info(self) -> dict:
+        """Get current database schema for inspection."""
+        pass
+
 
 class BdCliClient(BdClientBase):
     """Client for calling bd CLI commands and parsing JSON output."""
@@ -575,6 +585,28 @@ class BdCliClient(BdClientBase):
 
         return [BlockedIssue.model_validate(issue) for issue in data]
 
+    async def inspect_migration(self) -> dict:
+        """Get migration plan and database state for agent analysis.
+
+        Returns:
+            Migration plan dict with registered_migrations, warnings, etc.
+        """
+        data = await self._run_command("migrate", "--inspect")
+        if not isinstance(data, dict):
+            raise BdCommandError("Invalid response for inspect_migration")
+        return data
+
+    async def get_schema_info(self) -> dict:
+        """Get current database schema for inspection.
+
+        Returns:
+            Schema info dict with tables, version, config, sample IDs, etc.
+        """
+        data = await self._run_command("info", "--schema")
+        if not isinstance(data, dict):
+            raise BdCommandError("Invalid response for get_schema_info")
+        return data
+
     async def init(self, params: InitParams | None = None) -> str:
         """Initialize bd in current directory.
 

integrations/beads-mcp/src/beads_mcp/bd_daemon_client.py

@@ -430,6 +430,28 @@ class BdDaemonClient(BdClientBase):
         # This is a placeholder for when it's added
         raise NotImplementedError("Blocked operation not yet supported via daemon")
 
+    async def inspect_migration(self) -> dict:
+        """Get migration plan and database state for agent analysis.
+
+        Returns:
+            Migration plan dict with registered_migrations, warnings, etc.
+
+        Note:
+            This falls back to CLI since migrations are rare operations
+        """
+        raise NotImplementedError("inspect_migration not supported via daemon - use CLI client")
+
+    async def get_schema_info(self) -> dict:
+        """Get current database schema for inspection.
+
+        Returns:
+            Schema info dict with tables, version, config, sample IDs, etc.
+
+        Note:
+            This falls back to CLI since schema inspection is a rare operation
+        """
+        raise NotImplementedError("get_schema_info not supported via daemon - use CLI client")
+
     async def add_dependency(self, params: AddDependencyParams) -> None:
         """Add a dependency between issues.
 

integrations/beads-mcp/src/beads_mcp/server.py

@@ -18,7 +18,9 @@ from beads_mcp.tools import (
     beads_blocked,
     beads_close_issue,
     beads_create_issue,
+    beads_get_schema_info,
     beads_init,
+    beads_inspect_migration,
     beads_list_issues,
     beads_quickstart,
     beads_ready_work,
@@ -512,6 +514,39 @@ async def debug_env(workspace_root: str | None = None) -> str:
     return "".join(info)
 
 
+@mcp.tool(
+    name="inspect_migration",
+    description="Get migration plan and database state for agent analysis.",
+)
+@with_workspace
+async def inspect_migration(workspace_root: str | None = None) -> dict:
+    """Get migration plan and database state for agent analysis.
+    
+    AI agents should:
+    1. Review registered_migrations to understand what will run
+    2. Check warnings array for issues (missing config, version mismatch)
+    3. Verify missing_config is empty before migrating
+    4. Check invariants_to_check to understand safety guarantees
+    
+    Returns migration plan, current db state, warnings, and invariants.
+    """
+    return await beads_inspect_migration()
+
+
+@mcp.tool(
+    name="get_schema_info",
+    description="Get current database schema for inspection.",
+)
+@with_workspace
+async def get_schema_info(workspace_root: str | None = None) -> dict:
+    """Get current database schema for inspection.
+    
+    Returns tables, schema version, config, sample issue IDs, and detected prefix.
+    Useful for verifying database state before migrations.
+    """
+    return await beads_get_schema_info()
+
+
 async def async_main() -> None:
     """Async entry point for the MCP server."""
     await mcp.run_async(transport="stdio")

integrations/beads-mcp/src/beads_mcp/tools.py

@@ -453,6 +453,31 @@ async def beads_blocked() -> list[BlockedIssue]:
     return await client.blocked()
 
 
+async def beads_inspect_migration() -> dict:
+    """Get migration plan and database state for agent analysis.
+    
+    AI agents should:
+    1. Review registered_migrations to understand what will run
+    2. Check warnings array for issues (missing config, version mismatch)
+    3. Verify missing_config is empty before migrating
+    4. Check invariants_to_check to understand safety guarantees
+    
+    Returns migration plan, current db state, warnings, and invariants.
+    """
+    client = await _get_client()
+    return await client.inspect_migration()
+
+
+async def beads_get_schema_info() -> dict:
+    """Get current database schema for inspection.
+    
+    Returns tables, schema version, config, sample issue IDs, and detected prefix.
+    Useful for verifying database state before migrations.
+    """
+    client = await _get_client()
+    return await client.get_schema_info()
+
+
 async def beads_init(
     prefix: Annotated[str | None, "Issue prefix (e.g., 'myproject' for myproject-1, myproject-2)"] = None,
 ) -> str: