dss/dss/mcp/integrations/figma.py

"""
Figma Integration for MCP

Provides Figma API tools through circuit breaker pattern.
"""

import httpx
from typing import Dict, Any, List, Optional
from mcp import types

from .base import BaseIntegration
from ..config import integration_config


# Figma MCP Tool Definitions
FIGMA_TOOLS = [
    types.Tool(
        name="figma_get_file",
        description="Get Figma file metadata and structure",
        inputSchema={
            "type": "object",
            "properties": {
                "file_key": {
                    "type": "string",
                    "description": "Figma file key"
                }
            },
            "required": ["file_key"]
        }
    ),
    types.Tool(
        name="figma_get_styles",
        description="Get design styles (colors, text, effects) from Figma file",
        inputSchema={
            "type": "object",
            "properties": {
                "file_key": {
                    "type": "string",
                    "description": "Figma file key"
                }
            },
            "required": ["file_key"]
        }
    ),
    types.Tool(
        name="figma_get_components",
        description="Get component definitions from Figma file",
        inputSchema={
            "type": "object",
            "properties": {
                "file_key": {
                    "type": "string",
                    "description": "Figma file key"
                }
            },
            "required": ["file_key"]
        }
    ),
    types.Tool(
        name="figma_extract_tokens",
        description="Extract design tokens (variables) from Figma file",
        inputSchema={
            "type": "object",
            "properties": {
                "file_key": {
                    "type": "string",
                    "description": "Figma file key"
                }
            },
            "required": ["file_key"]
        }
    ),
    types.Tool(
        name="figma_get_node",
        description="Get specific node/component by ID from Figma file",
        inputSchema={
            "type": "object",
            "properties": {
                "file_key": {
                    "type": "string",
                    "description": "Figma file key"
                },
                "node_id": {
                    "type": "string",
                    "description": "Node ID to fetch"
                }
            },
            "required": ["file_key", "node_id"]
        }
    )
]


class FigmaIntegration(BaseIntegration):
    """Figma API integration with circuit breaker"""

    FIGMA_API_BASE = "https://api.figma.com/v1"

    def __init__(self, config: Dict[str, Any]):
        """
        Initialize Figma integration.

        Args:
            config: Must contain 'api_token' or use FIGMA_TOKEN from env
        """
        super().__init__("figma", config)
        self.api_token = config.get("api_token") or integration_config.FIGMA_TOKEN

        if not self.api_token:
            raise ValueError("Figma API token not configured")

        self.headers = {
            "X-Figma-Token": self.api_token
        }

    async def get_file(self, file_key: str) -> Dict[str, Any]:
        """
        Get Figma file metadata and structure.

        Args:
            file_key: Figma file key

        Returns:
            File data
        """
        async def _fetch():
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    f"{self.FIGMA_API_BASE}/files/{file_key}",
                    headers=self.headers,
                    timeout=30.0
                )
                response.raise_for_status()
                return response.json()

        return await self.call_api(_fetch)

    async def get_styles(self, file_key: str) -> Dict[str, Any]:
        """
        Get all styles from Figma file.

        Args:
            file_key: Figma file key

        Returns:
            Styles data
        """
        async def _fetch():
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    f"{self.FIGMA_API_BASE}/files/{file_key}/styles",
                    headers=self.headers,
                    timeout=30.0
                )
                response.raise_for_status()
                return response.json()

        return await self.call_api(_fetch)

    async def get_components(self, file_key: str) -> Dict[str, Any]:
        """
        Get all components from Figma file.

        Args:
            file_key: Figma file key

        Returns:
            Components data
        """
        async def _fetch():
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    f"{self.FIGMA_API_BASE}/files/{file_key}/components",
                    headers=self.headers,
                    timeout=30.0
                )
                response.raise_for_status()
                return response.json()

        return await self.call_api(_fetch)

    async def extract_tokens(self, file_key: str) -> Dict[str, Any]:
        """
        Extract design tokens (variables) from Figma file.

        Args:
            file_key: Figma file key

        Returns:
            Variables/tokens data
        """
        async def _fetch():
            async with httpx.AsyncClient() as client:
                # Get local variables
                response = await client.get(
                    f"{self.FIGMA_API_BASE}/files/{file_key}/variables/local",
                    headers=self.headers,
                    timeout=30.0
                )
                response.raise_for_status()
                return response.json()

        return await self.call_api(_fetch)

    async def get_node(self, file_key: str, node_id: str) -> Dict[str, Any]:
        """
        Get specific node from Figma file.

        Args:
            file_key: Figma file key
            node_id: Node ID

        Returns:
            Node data
        """
        async def _fetch():
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    f"{self.FIGMA_API_BASE}/files/{file_key}/nodes",
                    headers=self.headers,
                    params={"ids": node_id},
                    timeout=30.0
                )
                response.raise_for_status()
                return response.json()

        return await self.call_api(_fetch)


class FigmaTools:
    """MCP tool executor for Figma integration"""

    def __init__(self, config: Dict[str, Any]):
        """
        Args:
            config: Figma configuration (with api_token)
        """
        self.figma = FigmaIntegration(config)

    async def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """Execute Figma tool"""
        handlers = {
            "figma_get_file": self.figma.get_file,
            "figma_get_styles": self.figma.get_styles,
            "figma_get_components": self.figma.get_components,
            "figma_extract_tokens": self.figma.extract_tokens,
            "figma_get_node": self.figma.get_node
        }

        handler = handlers.get(tool_name)
        if not handler:
            return {"error": f"Unknown Figma tool: {tool_name}"}

        try:
            # Remove tool-specific prefix from arguments if needed
            clean_args = {k: v for k, v in arguments.items() if not k.startswith("_")}
            result = await handler(**clean_args)
            return result
        except Exception as e:
            return {"error": str(e)}