1mcp-seo/
2├── src/
3│   ├── handlers/
4│   │   ├── onpage_analysis.py
5│   │   ├── keyword_research.py
6│   │   └── serp_analysis.py
7│   ├── clients/
8│   │   └── dataforseo.py
9│   ├── cache.py
10│   └── server.py
11└── tests/
12    └── test_handlers.py

1# src/clients/dataforseo.py
2import aiohttp
3from typing import Dict, List
4
5class DataForSEOClient:
6    def __init__(self, login: str, password: str):
7        self.auth = aiohttp.BasicAuth(login, password)
8        self.base_url = "https://api.dataforseo.com/v3"
9
10    async def keyword_research(self, keywords: List[str]) -> Dict:
11        async with aiohttp.ClientSession(auth=self.auth) as session:
12            payload = {
13                "keywords": keywords,
14                "language_code": "fr",
15                "location_code": 2250  # France
16            }
17            async with session.post(f"{self.base_url}/keywords_data", json=payload) as response:
18                return await response.json()
19
20    async def serp_analysis(self, keyword: str) -> Dict:
21        async with aiohttp.ClientSession(auth=self.auth) as session:
22            async with session.get(f"{self.base_url}/serp/{keyword}") as response:
23                return await response.json()

1# src/handlers/onpage_analysis.py
2from dataclasses import dataclass
3from typing import List, Dict
4
5@dataclass
6class OnPageResult:
7    title_length: int
8    meta_description: str
9    headers_structure: Dict[str, int]
10    performance_score: float
11
12class OnPageHandler:
13    def analyze(self, url: str) -> OnPageResult:
14        # Simulation d'un onpage analysis complet
15        title = self._get_title(url)
16        meta = self._get_meta_description(url)
17        headers = self._analyze_headers(url)
18        performance = self._get_lighthouse_score(url)
19
20        return OnPageResult(
21            title_length=len(title),
22            meta_description=meta,
23            headers_structure=headers,
24            performance_score=performance
25        )
26
27    def _get_title(self, url: str) -> str:
28        # Logique d'extraction du titre
29        pass
30
31    def _get_meta_description(self, url: str) -> str:
32        # Logique d'extraction meta description
33        pass
34
35    def _analyze_headers(self, url: str) -> Dict[str, int]:
36        # Analyse de la structure des headers
37        return {
38            "h1": 1,
39            "h2": 3,
40            "h3": 5
41        }
42
43    def _get_lighthouse_score(self, url: str) -> float:
44        # Simulation score Lighthouse
45        return 94.5

1# src/handlers/keyword_research.py
2class KeywordResearchHandler:
3    def __init__(self, dataforseo_client):
4        self.client = dataforseo_client
5
6    async def research(self, seed_keyword: str) -> List[Dict]:
7        results = await self.client.keyword_research([seed_keyword])
8
9        # Filtrage et analyse des résultats
10        keywords = []
11        for result in results:
12            keywords.append({
13                "keyword": result["keyword"],
14                "volume": result["search_volume"],
15                "cpc": result["cpc"],
16                "competition": result["competition"]
17            })
18
19        return sorted(keywords, key=lambda x: x["volume"], reverse=True)

1# src/server.py
2import asyncio
3from typing import Dict, Any
4
5class MCPSEOServer:
6    def __init__(self, dataforseo_client):
7        self.onpage_handler = OnPageHandler()
8        self.keyword_handler = KeywordResearchHandler(dataforseo_client)
9        self.cache = {}
10
11    async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
12        """
13        Dispatche les requêtes vers les bons handlers
14        Avec cache intelligent (15 min)
15        """
16        cache_key = self._generate_cache_key(request)
17
18        if cache_key in self.cache:
19            return self.cache[cache_key]
20
21        result = None
22        match request["type"]:
23            case "onpage_analysis":
24                result = self.onpage_handler.analyze(request["url"])
25            case "keyword_research":
26                result = await self.keyword_handler.research(request["keyword"])
27
28        # Cache avec expiration
29        self.cache[cache_key] = result
30        asyncio.create_task(self._expire_cache(cache_key, 900))  # 15 min
31
32        return result
33
34    def _generate_cache_key(self, request):
35        # Génération d'une clé unique basée sur la requête
36        pass
37
38    async def _expire_cache(self, key, ttl):
39        await asyncio.sleep(ttl)
40        self.cache.pop(key, None)