De la comptabilité à la prospection commerciale, du reporting à l'analyse sémantique des données — le blueprint complet pour construire l'entreprise du futur où l'IA orchestre chaque processus.
Imaginez une entreprise où chaque processus répétitif — de la saisie comptable à la génération de rapports, de la prospection commerciale au support client — est orchestré par une intelligence artificielle. Pas une utopie futuriste : un blueprint actionnable que vous pouvez commencer à construire aujourd'hui avec Claude Code.
La Partie 1 de ce guide vous a enseigné les fondamentaux : MCP, Hooks, Skills, Scripting headless, Subagents. Cette Partie 2 passe à l'échelle. Nous allons construire, brique par brique, une entreprise autonome où Claude Code est le cerveau central qui connecte, analyse, décide et exécute.
Le diagramme ci-dessous représente l'architecture cible. Claude Code se positionne au centre, connecté à tous les systèmes de l'entreprise via des MCP servers spécialisés, avec des hooks pour la gouvernance et des skills pour chaque workflow métier.
┌──────────────────────────────────────────┐
│ HUMAN-IN-THE-LOOP │
│ Approbations · Supervision · Stratégie │
└────────────────┬─────────────────────────┘
│
┌─────────────────────┼──────────────────────┐
│ ▼ │
┌─────────┐ ┌──────────────────────────────────────┐ ┌──────────┐
│ HOOKS │◀──│ CLAUDE CODE CENTRAL │──▶│ SKILLS │
│ │ │ │ │ │
│Validation│ │ Raisonnement · Décision · Actions │ │Comptable │
│Sécurité │ │ Orchestration de processus │ │Prospection│
│Audit │ │ Analyse sémantique des données │ │Reporting │
│Compliance│ │ Génération de rapports │ │Support │
└─────────┘ └──────┬────┬────┬────┬────┬──────────┘ │Juridique │
│ │ │ │ │ │RH │
▼ ▼ ▼ ▼ ▼ └──────────┘
┌────┐┌────┐┌────┐┌────┐┌────┐
│MCP ││MCP ││MCP ││MCP ││MCP │
│ ││ ││ ││ ││ │
│ERP ││CRM ││Bank││Data││Mail│
│API ││API ││API ││Whse││API │
└────┘└────┘└────┘└────┘└────┘
│ │ │ │ │
┌──┴─────┴─────┴─────┴─────┴──┐
│ SCHEDULED TASKS (Cron) │
│ Daily · Weekly · Triggers │
└──────────────────────────────┘
| Entreprise | Secteur | Automatisation IA | Résultat |
|---|---|---|---|
| Klarna | Fintech | Support client IA (remplacement 700 postes) | Résolution en 2 min vs 11 min, CSAT stable |
| Shopify | E-commerce | IA-first policy (tout passe par l'IA d'abord) | Productivité x3 sur les tâches administratives |
| Duolingo | EdTech | Génération de contenu IA à 80% | Délais de production divisés par 5 |
| Mercado Libre | Marketplace | Listings, fraud detection, routing IA | 1.4M listings générés/jour, fraude -60% |
| Ramp | Fintech | Comptabilité, notes de frais, rapports IA | Clôture comptable de 5j à 1j |
Ce guide suppose que vous avez lu la Partie 1 et que vous maîtrisez :
claude -p, --output-format json, pipingLe Semantic Layer est la pièce manquante entre l'IA et les données d'entreprise. Sans lui, Claude ne voit que des tables SQL brutes, des colonnes ambigües et des relations cachées. Avec lui, Claude comprend la signification métier de chaque donnée : ce qu'est le "MRR", comment calculer le "churn", quelle est la définition officielle du "revenu net".
Un Semantic Layer est une couche d'abstraction posée entre les données brutes (tables SQL) et les consommateurs (BI tools, LLMs, APIs). Il définit :
| Solution | Approche | Intégration IA | Cas d'usage |
|---|---|---|---|
| dbt Semantic Layer | YAML dans le projet dbt, MetricFlow engine | API REST/GraphQL, native dbt Cloud | Entreprises utilisant dbt (standard) |
| Cube.js | Fichiers JavaScript/YAML, moteur Cube Store | API REST, SDK, cache intégré | Startups, apps analytiques embarquées |
| Lightdash | dbt metrics natif, UI BI intégrée | Via dbt Semantic Layer | Teams data-first avec dbt |
| Looker (Google) | LookML, modèles sémantiques propriétaires | Gemini intégré, API LookML | Grandes entreprises GCP |
| AtScale | Virtual OLAP, compatible MDX | Connecteurs BI universels | Migration depuis OLAP legacy |
┌────────────┐ ┌─────────────┐ ┌──────────────┐ ┌──────────┐
│ Sources │────▶│ dbt Models │────▶│ Semantic │────▶│ Claude │
│ │ │ │ │ Layer │ │ Code │
│ PostgreSQL │ │ staging │ │ │ │ │
│ Stripe API │ │ intermediate│ │ Entités │ │ Queries │
│ HubSpot │ │ marts │ │ Mesures │ │ Insights │
│ Sheets │ │ │ │ Dimensions │ │ Rapports │
└────────────┘ └─────────────┘ └──────────────┘ └──────────┘
EL(T) Transform Sémantique Intelligence
Voici un modèle sémantique complet pour le revenu d'une entreprise SaaS, défini en YAML dans votre projet dbt :
semantic_models:
- name: revenue
description: "Modèle de revenu SaaS - MRR, ARR, churn"
model: ref('fct_subscriptions')
defaults:
agg_time_dimension: subscription_date
entities:
- name: subscription_id
type: primary
- name: customer_id
type: foreign
expr: customer_id
- name: plan_id
type: foreign
expr: plan_id
measures:
- name: total_mrr
description: "Monthly Recurring Revenue total"
expr: mrr_amount
agg: sum
- name: subscription_count
description: "Nombre d'abonnements actifs"
agg: count
- name: avg_contract_value
description: "Valeur moyenne d'un contrat"
expr: contract_value
agg: average
- name: churned_mrr
description: "MRR perdu (churn)"
expr: "case when status = 'churned' then mrr_amount else 0 end"
agg: sum
dimensions:
- name: customer_segment
type: categorical
expr: customer_segment
description: "Segment client : startup, mid-market, enterprise"
- name: plan_type
type: categorical
expr: plan_name
description: "Type de plan : free, pro, business, enterprise"
- name: subscription_date
type: time
expr: created_at
time_granularity: day
- name: region
type: categorical
expr: billing_country
description: "Région géographique du client"
metrics:
- name: mrr
description: "Monthly Recurring Revenue"
type: simple
type_params:
measure: total_mrr
- name: net_mrr_retention
description: "Net MRR Retention Rate"
type: derived
type_params:
expr: "(total_mrr - churned_mrr) / total_mrr * 100"
metrics:
- name: total_mrr
- name: churned_mrr
- name: arpu
description: "Average Revenue Per User"
type: derived
type_params:
expr: "total_mrr / subscription_count"
metrics:
- name: total_mrr
- name: subscription_count
- name: ltv_estimate
description: "Estimated Lifetime Value (MRR * avg_lifetime_months)"
type: derived
type_params:
expr: "(total_mrr / subscription_count) * 24"
metrics:
- name: total_mrr
- name: subscription_countL'objectif est de créer un serveur MCP que Claude Code peut interroger en langage naturel. Le serveur traduit les questions en requêtes sémantiques, exécute via MetricFlow ou Cube.js, et retourne les données structurées.
"""
MCP Server : Semantic Layer Bridge
Connecte Claude Code au dbt Semantic Layer via MetricFlow API
"""
import json
import httpx
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("semantic-layer")
METRICFLOW_URL = "http://localhost:8765/api/v1"
DBT_CLOUD_TOKEN = "dbt_cloud_token_here"
# Cache du catalogue de métriques
_metrics_catalog = None
@mcp.tool()
async def list_metrics() -> str:
"""Liste toutes les métriques disponibles dans le Semantic Layer.
Retourne nom, description, type et dimensions disponibles."""
global _metrics_catalog
if _metrics_catalog is None:
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{METRICFLOW_URL}/metrics",
headers={"Authorization": f"Bearer {DBT_CLOUD_TOKEN}"}
)
_metrics_catalog = resp.json()
result = []
for m in _metrics_catalog["metrics"]:
result.append({
"name": m["name"],
"description": m["description"],
"type": m["type"],
"dimensions": [d["name"] for d in m.get("dimensions", [])],
"time_grains": m.get("time_granularities", ["day", "week", "month"])
})
return json.dumps(result, indent=2)
@mcp.tool()
async def query_metric(
metric_names: list[str],
group_by: list[str] | None = None,
time_grain: str = "month",
time_range: str = "last_6_months",
filters: dict | None = None,
order_by: str | None = None,
limit: int = 100
) -> str:
"""Interroge une ou plusieurs métriques du Semantic Layer.
Args:
metric_names: Liste des métriques (ex: ["mrr", "arpu"])
group_by: Dimensions de regroupement (ex: ["customer_segment", "region"])
time_grain: Granularité temporelle (day, week, month, quarter, year)
time_range: Période (last_7_days, last_30_days, last_6_months, last_year, ytd)
filters: Filtres optionnels (ex: {"customer_segment": "enterprise"})
order_by: Colonne de tri (ex: "mrr DESC")
limit: Nombre max de résultats
Returns:
Données tabulaires au format JSON avec métadonnées
"""
payload = {
"metrics": [{"name": m} for m in metric_names],
"group_by": group_by or [],
"time_grain": time_grain,
"time_range": time_range,
"limit": limit
}
if filters:
payload["where"] = [
{"dimension": k, "operator": "=", "value": v}
for k, v in filters.items()
]
if order_by:
payload["order_by"] = [order_by]
async with httpx.AsyncClient(timeout=30) as client:
resp = await client.post(
f"{METRICFLOW_URL}/query",
json=payload,
headers={"Authorization": f"Bearer {DBT_CLOUD_TOKEN}"}
)
data = resp.json()
return json.dumps({
"query": payload,
"row_count": len(data.get("rows", [])),
"columns": data.get("columns", []),
"rows": data.get("rows", []),
"sql_generated": data.get("sql", ""),
"cache_hit": data.get("cache_hit", False)
}, indent=2)
@mcp.tool()
async def explain_metric(metric_name: str) -> str:
"""Retourne la définition complète d'une métrique :
formule SQL, dimensions, entités, description métier."""
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{METRICFLOW_URL}/metrics/{metric_name}",
headers={"Authorization": f"Bearer {DBT_CLOUD_TOKEN}"}
)
metric = resp.json()
return json.dumps({
"name": metric["name"],
"description": metric["description"],
"type": metric["type"],
"sql_expression": metric.get("expr", "N/A"),
"aggregation": metric.get("agg", "N/A"),
"base_model": metric.get("model", "N/A"),
"dimensions": metric.get("dimensions", []),
"filters_available": metric.get("filters", []),
"business_definition": metric.get("meta", {}).get("business_definition", ""),
"owner": metric.get("meta", {}).get("owner", "data-team"),
"last_updated": metric.get("meta", {}).get("last_updated", "")
}, indent=2)
@mcp.tool()
async def compare_periods(
metric_name: str,
current_period: str = "this_month",
comparison_period: str = "last_month",
group_by: list[str] | None = None
) -> str:
"""Compare une métrique entre deux périodes.
Retourne les valeurs, le delta absolu et le delta en pourcentage."""
# Query current period
current = await _query_period(metric_name, current_period, group_by)
previous = await _query_period(metric_name, comparison_period, group_by)
comparisons = []
for key in current:
curr_val = current[key]
prev_val = previous.get(key, 0)
delta = curr_val - prev_val
delta_pct = (delta / prev_val * 100) if prev_val != 0 else None
comparisons.append({
"dimension": key,
"current": curr_val,
"previous": prev_val,
"delta": delta,
"delta_percent": round(delta_pct, 2) if delta_pct else None,
"trend": "up" if delta > 0 else "down" if delta < 0 else "flat"
})
return json.dumps({
"metric": metric_name,
"current_period": current_period,
"comparison_period": comparison_period,
"comparisons": comparisons
}, indent=2)
if __name__ == "__main__":
mcp.run(transport="stdio"){
"mcpServers": {
"semantic-layer": {
"command": "python",
"args": ["/path/to/mcp-semantic-layer/server.py"],
"env": {
"DBT_CLOUD_TOKEN": "${DBT_CLOUD_TOKEN}",
"METRICFLOW_URL": "http://localhost:8765/api/v1"
}
}
}
}Voici comment une question en français se transforme en insight actionable :
$ claude -p "Quel est notre MRR par segment client ce trimestre vs le précédent ?
Identifie les segments en croissance et ceux en déclin.
Génère un rapport avec recommandations."Claude exécute automatiquement les étapes suivantes :
Cube.js offre une alternative puissante avec un moteur de cache intégré (Cube Store) et une API REST native. Voici comment définir un schéma sémantique Cube.js et le connecter à Claude :
cube('Revenue', {
sql_table: 'public.fct_subscriptions',
measures: {
mrr: {
type: 'sum',
sql: 'mrr_amount',
description: 'Monthly Recurring Revenue',
format: 'currency'
},
customerCount: {
type: 'countDistinct',
sql: 'customer_id',
description: 'Nombre de clients uniques'
},
arpu: {
type: 'number',
sql: `${mrr} / NULLIF(${customerCount}, 0)`,
description: 'Average Revenue Per User'
},
churnRate: {
type: 'number',
sql: `COUNT(CASE WHEN status = 'churned' THEN 1 END)::float
/ NULLIF(COUNT(*), 0) * 100`,
description: 'Taux de churn mensuel en %',
format: 'percent'
}
},
dimensions: {
segment: {
type: 'string',
sql: 'customer_segment',
description: 'Segment : startup, mid-market, enterprise'
},
plan: {
type: 'string',
sql: 'plan_name',
description: 'Plan tarifaire'
},
region: {
type: 'string',
sql: 'billing_country',
description: 'Pays de facturation'
},
createdAt: {
type: 'time',
sql: 'created_at',
description: 'Date de création de l\'abonnement'
}
},
preAggregations: {
mrrBySegment: {
measures: [mrr, customerCount],
dimensions: [segment],
timeDimension: createdAt,
granularity: 'month',
refreshKey: { every: '1 hour' }
}
}
});"""MCP Server pour Cube.js"""
import httpx
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("cubejs")
CUBE_API = "http://localhost:4000/cubejs-api/v1"
CUBE_TOKEN = "your_cube_jwt_token"
@mcp.tool()
async def cube_query(
measures: list[str],
dimensions: list[str] | None = None,
time_dimension: str | None = None,
date_range: list[str] | None = None,
filters: list[dict] | None = None,
order: dict | None = None,
limit: int = 100
) -> str:
"""Interroge l'API Cube.js avec des mesures et dimensions."""
query = {"measures": measures, "limit": limit}
if dimensions:
query["dimensions"] = dimensions
if time_dimension and date_range:
query["timeDimensions"] = [{
"dimension": time_dimension,
"dateRange": date_range,
"granularity": "month"
}]
if filters:
query["filters"] = filters
if order:
query["order"] = order
async with httpx.AsyncClient(timeout=30) as client:
resp = await client.post(
f"{CUBE_API}/load",
json={"query": query},
headers={"Authorization": CUBE_TOKEN}
)
return resp.text
if __name__ == "__main__":
mcp.run(transport="stdio")mrr_amount n'est plus un chiffre ambigu — c'est le "Monthly Recurring Revenue, somme des abonnements actifs, excluant les périodes d'essai, en devise de facturation". Cette précision élimine les hallucinations sur les données.
| Pratique | Pourquoi | Exemple |
|---|---|---|
| Descriptions exhaustives | Le LLM utilise la description pour comprendre quand utiliser la métrique | description: "MRR excluant les trials, en EUR" |
| Nommage cohérent | Évite les confusions (revenue vs mrr vs arr) | Convention entity_measure |
| Méta-données métier | Enrichit le contexte pour l'IA | meta: { owner: "finance", sla: "daily" } |
| Tests unitaires | Garantit l'intégrité des calculs | dbt test --select metric:mrr |
| Versionning | Traçabilité des changements de définition | Git + dbt docs + changelog |
La comptabilité est le candidat idéal pour l'automatisation IA : des processus répétitifs, des règles bien définies, un volume élevé de documents, et un coût d'erreur significatif. Un comptable passe en moyenne 60% de son temps sur des tâches automatisables : saisie de factures, rapprochement bancaire, classification de dépenses.
Le workflow complet, de la réception d'un email à la comptabilisation :
Email reçu OCR / Extraction Vérification Comptabilisation
┌──────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Facture │─────▶│ Claude Code │─────▶│ Matching PO │─────▶│ Journal │
│ PDF/IMG │ │ extrait : │ │ Contrôle TVA │ │ Entry draft │
│ │ │ - Fournisseur│ │ Détection │ │ │
│ │ │ - Montant HT │ │ anomalies │ │ ──▶ HUMAIN │
│ │ │ - TVA │ │ (doublons, │ │ VALIDE │
│ │ │ - Lignes │ │ montant) │ │ │
│ │ │ - Date │ │ │ │ ──▶ Posting │
└──────────┘ └──────────────┘ └──────────────┘ └──────────────┘
"""
MCP Server : Accounting Bridge
Connecte Claude Code à Xero ou QuickBooks pour la comptabilité automatisée.
"""
import json
from datetime import datetime, timedelta
from mcp.server.fastmcp import FastMCP
import httpx
mcp = FastMCP("accounting")
XERO_API = "https://api.xero.com/api.xro/2.0"
XERO_TENANT_ID = ""
XERO_TOKEN = ""
headers = {
"Authorization": f"Bearer {XERO_TOKEN}",
"Xero-Tenant-Id": XERO_TENANT_ID,
"Content-Type": "application/json"
}
@mcp.tool()
async def extract_invoice_data(file_path: str) -> str:
"""Extrait les données d'une facture PDF/image via OCR + Claude.
Retourne : fournisseur, montant HT, TVA, lignes, date, numéro."""
# Claude lit naturellement les PDFs/images
# Le hook PreToolUse vérifie que le fichier est bien une facture
return json.dumps({
"status": "extracted",
"instructions": "Utilisez l'outil Read pour lire le fichier, "
"puis structurez les données extraites."
})
@mcp.tool()
async def search_chart_of_accounts(query: str) -> str:
"""Recherche dans le plan comptable pour classifier une dépense.
Ex: 'fournitures bureau' -> compte 6064."""
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{XERO_API}/Accounts",
headers=headers,
params={"where": f'Name.Contains("{query}")'}
)
accounts = resp.json().get("Accounts", [])
return json.dumps([{
"code": a["Code"],
"name": a["Name"],
"type": a["Type"],
"tax_type": a.get("TaxType", "N/A")
} for a in accounts[:10]], indent=2)
@mcp.tool()
async def create_invoice_draft(
vendor_name: str,
invoice_number: str,
date: str,
due_date: str,
line_items: list[dict],
currency: str = "EUR",
notes: str = ""
) -> str:
"""Crée un BROUILLON de facture fournisseur dans Xero.
IMPORTANT : Statut DRAFT uniquement, nécessite validation humaine.
Args:
line_items: [{"description": str, "quantity": float,
"unit_price": float, "account_code": str,
"tax_type": str}]
"""
payload = {
"Type": "ACCPAY",
"Contact": {"Name": vendor_name},
"InvoiceNumber": invoice_number,
"Date": date,
"DueDate": due_date,
"Status": "DRAFT", # TOUJOURS DRAFT
"CurrencyCode": currency,
"LineItems": [{
"Description": item["description"],
"Quantity": item["quantity"],
"UnitAmount": item["unit_price"],
"AccountCode": item["account_code"],
"TaxType": item.get("tax_type", "INPUT")
} for item in line_items]
}
if notes:
payload["Reference"] = notes
async with httpx.AsyncClient() as client:
resp = await client.post(
f"{XERO_API}/Invoices",
headers=headers,
json={"Invoices": [payload]}
)
result = resp.json()
invoice_id = result["Invoices"][0]["InvoiceID"]
return json.dumps({
"status": "DRAFT_CREATED",
"invoice_id": invoice_id,
"message": f"Facture {invoice_number} créée en BROUILLON. "
f"Validation humaine requise avant approbation.",
"approval_url": f"https://go.xero.com/AccountsPayable/{invoice_id}"
}, indent=2)
@mcp.tool()
async def match_purchase_order(
vendor_name: str,
amount: float,
tolerance: float = 0.05
) -> str:
"""Recherche un bon de commande correspondant à la facture.
Tolérance par défaut : 5% sur le montant."""
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{XERO_API}/PurchaseOrders",
headers=headers,
params={
"where": f'Contact.Name=="{vendor_name}" '
f'AND Status=="AUTHORISED"',
"order": "Date DESC"
}
)
orders = resp.json().get("PurchaseOrders", [])
matches = []
for po in orders:
po_amount = float(po.get("Total", 0))
diff = abs(po_amount - amount) / po_amount if po_amount else 1
if diff <= tolerance:
matches.append({
"po_number": po["PurchaseOrderNumber"],
"amount": po_amount,
"date": po["Date"],
"match_confidence": round((1 - diff) * 100, 1)
})
return json.dumps({
"vendor": vendor_name,
"invoice_amount": amount,
"matches_found": len(matches),
"matches": sorted(matches, key=lambda x: x["match_confidence"], reverse=True)
}, indent=2)
@mcp.tool()
async def detect_anomalies(invoice_data: dict) -> str:
"""Détecte les anomalies dans une facture :
- Doublon (même numéro, même fournisseur)
- Montant inhabituel (> 2 écarts-types de la moyenne fournisseur)
- Date suspecte (antérieure à 90 jours)
- TVA incorrecte (taux non standard)"""
anomalies = []
# Vérifier doublons
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{XERO_API}/Invoices",
headers=headers,
params={
"where": f'InvoiceNumber=="{invoice_data["number"]}" '
f'AND Contact.Name=="{invoice_data["vendor"]}"'
}
)
existing = resp.json().get("Invoices", [])
if existing:
anomalies.append({
"type": "DUPLICATE",
"severity": "HIGH",
"message": f"Facture {invoice_data['number']} existe déjà pour {invoice_data['vendor']}"
})
# Vérifier montant
if invoice_data.get("amount", 0) > 50000:
anomalies.append({
"type": "HIGH_AMOUNT",
"severity": "MEDIUM",
"message": f"Montant élevé ({invoice_data['amount']}€) - vérification manuelle recommandée"
})
# Vérifier date
inv_date = datetime.strptime(invoice_data.get("date", ""), "%Y-%m-%d")
if (datetime.now() - inv_date).days > 90:
anomalies.append({
"type": "OLD_INVOICE",
"severity": "LOW",
"message": f"Facture datée de plus de 90 jours ({invoice_data['date']})"
})
return json.dumps({
"anomalies_count": len(anomalies),
"anomalies": anomalies,
"recommendation": "BLOCK" if any(a["severity"] == "HIGH" for a in anomalies)
else "REVIEW" if anomalies else "OK"
}, indent=2)
@mcp.tool()
async def bank_reconciliation(
account_id: str,
start_date: str,
end_date: str
) -> str:
"""Rapprochement bancaire automatique.
Compare les transactions bancaires aux factures et suggère les correspondances."""
async with httpx.AsyncClient() as client:
# Récupérer les transactions bancaires
bank_resp = await client.get(
f"{XERO_API}/BankTransactions",
headers=headers,
params={"where": f'BankAccount.AccountID==guid("{account_id}") '
f'AND Date>=DateTime({start_date.replace("-", ",")}) '
f'AND Date<=DateTime({end_date.replace("-", ",")})',
"order": "Date"}
)
transactions = bank_resp.json().get("BankTransactions", [])
# Récupérer les factures en attente
inv_resp = await client.get(
f"{XERO_API}/Invoices",
headers=headers,
params={"where": 'Status=="AUTHORISED"', "order": "Date"}
)
invoices = inv_resp.json().get("Invoices", [])
# Matching automatique
matches = []
unmatched_txns = []
for txn in transactions:
matched = False
for inv in invoices:
if abs(float(txn["Total"]) - float(inv["Total"])) < 0.01:
matches.append({
"transaction": txn["BankTransactionID"],
"invoice": inv["InvoiceNumber"],
"amount": txn["Total"],
"confidence": "HIGH"
})
matched = True
break
if not matched:
unmatched_txns.append({
"id": txn["BankTransactionID"],
"date": txn["Date"],
"amount": txn["Total"],
"reference": txn.get("Reference", "")
})
return json.dumps({
"period": f"{start_date} to {end_date}",
"total_transactions": len(transactions),
"auto_matched": len(matches),
"unmatched": len(unmatched_txns),
"match_rate": f"{len(matches)/len(transactions)*100:.1f}%" if transactions else "0%",
"matches": matches[:20],
"unmatched_transactions": unmatched_txns[:20]
}, indent=2)
if __name__ == "__main__":
mcp.run(transport="stdio")# /process-invoice
Traitement automatisé d'une facture fournisseur.
## Workflow
1. **Lire** le fichier PDF/image de la facture (outil Read)
2. **Extraire** : fournisseur, numéro, date, montant HT, TVA, lignes
3. **Vérifier les anomalies** : detect_anomalies()
- Si BLOCK : arrêter et notifier le comptable
- Si REVIEW : signaler mais continuer
4. **Classifier** chaque ligne contre le plan comptable : search_chart_of_accounts()
5. **Matcher** avec un bon de commande : match_purchase_order()
6. **Créer le brouillon** : create_invoice_draft()
7. **Notifier** le comptable pour validation (Slack/email)
## Règles
- TOUJOURS créer en DRAFT, jamais en AUTHORISED
- Montant > 10 000€ : double validation obligatoire
- Nouveau fournisseur : signaler pour vérification KYC
- TVA à 0% : vérifier autoliquidation ou exonération
## Output
JSON structuré avec :
- Statut (CREATED, BLOCKED, NEEDS_REVIEW)
- ID de la facture brouillon
- Anomalies détectées
- Confiance du matching PO
- Lien direct vers Xero pour validation#!/bin/bash
# Clôture mensuelle automatisée avec Claude Code
# Exécuté le 1er de chaque mois à 8h00
MONTH=$(date -d "yesterday" +%Y-%m)
START="${MONTH}-01"
END=$(date -d "yesterday" +%Y-%m-%d)
echo "=== CLÔTURE MENSUELLE ${MONTH} ==="
# Étape 1 : Rapprochement bancaire
claude -p "Exécute le rapprochement bancaire pour la période ${START} à ${END}.
Utilise bank_reconciliation() pour chaque compte.
Signale toutes les transactions non rapprochées.
Génère un rapport de rapprochement." \
--output-format json > /tmp/reconciliation.json
# Étape 2 : Vérification des écritures en attente
claude -p "Vérifie qu'il n'y a aucune facture DRAFT en attente pour ${MONTH}.
Liste toutes les factures non validées.
Vérifie les provisions manquantes (loyer, salaires, impôts)." \
--output-format json > /tmp/pending-check.json
# Étape 3 : Génération du P&L et bilan
claude -p "Génère le P&L et le bilan pour ${MONTH}.
Compare avec le mois précédent et le budget.
Identifie les écarts significatifs (>10%).
Format : rapport HTML avec tableaux et graphiques." \
--output-format json > /tmp/financial-report.json
# Étape 4 : Notification
claude -p "Envoie le rapport de clôture mensuelle ${MONTH} par email au CFO.
Inclus : rapprochement, factures en attente, P&L, bilan.
Sujet : 'Clôture ${MONTH} - Rapport automatisé'" \
--output-format json
echo "=== CLÔTURE ${MONTH} TERMINÉE ==="| Processus | Avant (manuel) | Après (Claude) | Gain |
|---|---|---|---|
| Saisie facture | 12 min/facture | 45 sec/facture | -94% |
| Classification | 5 min + erreurs | Automatique, 97% précision | -100% |
| Rapprochement bancaire | 4h/mois | 15 min/mois | -94% |
| Clôture mensuelle | 5 jours | 1 jour | -80% |
| Détection anomalies | Spot-check (5%) | 100% des factures | Couverture totale |
| Prévision trésorerie | Excel mensuel | Rolling forecast quotidien | Temps réel |
La prospection commerciale est le moteur de croissance de toute entreprise. Mais c'est aussi l'un des processus les plus chronophages et les moins efficaces : un commercial passe en moyenne 21% de son temps à vendre effectivement. Le reste ? Recherche de leads, enrichissement de données, rédaction d'emails, relances, saisie CRM.
Claude Code peut automatiser 80% de ce travail non-commercial, transformant chaque vendeur en machine de vente assistée par IA.
Identification Enrichissement Scoring Outreach Follow-up
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ Sources │──────▶│ Claude │──────▶│ ICP │───▶│ Email │─────▶│ Séquence │
│ │ │ enrichit │ │ Score │ │ perso │ │ auto │
│ LinkedIn │ │ │ │ │ │ │ │ │
│ Apollo │ │ Firmo │ │ Fit: 87% │ │ Objet │ │ J+3 rlc │
│ Clearbit │ │ Techno │ │ Intent:72│ │ Corps │ │ J+7 rlc │
│ Web scrap│ │ Intent │ │ Total:82 │ │ CTA │ │ J+14 call│
└──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘
│
┌────┴────┐
│ HUMAIN │
│ VALIDE │
│ avant │
│ envoi │
└─────────┘
"""
MCP Server : Sales Prospection Engine
Enrichissement, scoring, génération d'outreach, sync CRM.
"""
import json
import httpx
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("prospection")
APOLLO_API = "https://api.apollo.io/v1"
APOLLO_KEY = ""
HUBSPOT_API = "https://api.hubapi.com/crm/v3"
HUBSPOT_TOKEN = ""
@mcp.tool()
async def enrich_company(
domain: str
) -> str:
"""Enrichit les données d'une entreprise à partir de son domaine.
Retourne : taille, secteur, technologies, funding, décideurs clés."""
async with httpx.AsyncClient() as client:
resp = await client.post(
f"{APOLLO_API}/organizations/enrich",
headers={"X-Api-Key": APOLLO_KEY},
json={"domain": domain}
)
org = resp.json().get("organization", {})
return json.dumps({
"name": org.get("name"),
"domain": domain,
"industry": org.get("industry"),
"employee_count": org.get("estimated_num_employees"),
"revenue_range": org.get("annual_revenue_printed"),
"technologies": org.get("technologies", [])[:20],
"funding_total": org.get("total_funding_printed"),
"last_funding": org.get("last_funding_round", {}),
"linkedin_url": org.get("linkedin_url"),
"location": {
"city": org.get("city"),
"country": org.get("country")
},
"keywords": org.get("keywords", [])[:10]
}, indent=2)
@mcp.tool()
async def find_decision_makers(
domain: str,
titles: list[str] = None,
limit: int = 5
) -> str:
"""Trouve les décideurs clés dans une entreprise.
Filtre par titre (CEO, CTO, VP Sales, etc.)."""
if titles is None:
titles = ["CEO", "CTO", "CFO", "VP", "Director", "Head of"]
async with httpx.AsyncClient() as client:
resp = await client.post(
f"{APOLLO_API}/mixed_people/search",
headers={"X-Api-Key": APOLLO_KEY},
json={
"organization_domains": [domain],
"person_titles": titles,
"page": 1,
"per_page": limit
}
)
people = resp.json().get("people", [])
return json.dumps([{
"name": p.get("name"),
"title": p.get("title"),
"email": p.get("email"),
"linkedin_url": p.get("linkedin_url"),
"department": p.get("departments", ["Unknown"])[0],
"seniority": p.get("seniority")
} for p in people], indent=2)
@mcp.tool()
async def score_lead(
company_data: dict,
icp_criteria: dict = None
) -> str:
"""Score un lead selon le profil client idéal (ICP).
Args:
company_data: Données enrichies de l'entreprise
icp_criteria: Critères ICP personnalisés
ICP par défaut (SaaS B2B) :
- Taille : 50-500 employés (30 pts)
- Secteur : Tech, Finance, Healthcare (20 pts)
- Technologies : utilise des outils modernes (15 pts)
- Funding : Série A+ (15 pts)
- Géo : Europe / Amérique du Nord (10 pts)
- Intent : signaux d'achat détectés (10 pts)
"""
if icp_criteria is None:
icp_criteria = {
"employee_range": [50, 500],
"target_industries": ["Technology", "Financial Services", "Healthcare"],
"target_technologies": ["AWS", "Python", "React", "Kubernetes"],
"min_funding": "Series A",
"target_geos": ["US", "UK", "FR", "DE"]
}
score = 0
breakdown = {}
# Taille (30 pts)
emp = company_data.get("employee_count", 0)
if icp_criteria["employee_range"][0] <= emp <= icp_criteria["employee_range"][1]:
score += 30
breakdown["company_size"] = {"score": 30, "max": 30, "detail": f"{emp} employés"}
elif emp > 0:
size_score = max(0, 30 - abs(emp - 250) / 25)
score += size_score
breakdown["company_size"] = {"score": round(size_score), "max": 30, "detail": f"{emp} employés"}
# Secteur (20 pts)
industry = company_data.get("industry", "")
if industry in icp_criteria["target_industries"]:
score += 20
breakdown["industry"] = {"score": 20, "max": 20, "detail": industry}
else:
breakdown["industry"] = {"score": 0, "max": 20, "detail": industry}
# Technologies (15 pts)
techs = set(company_data.get("technologies", []))
target_techs = set(icp_criteria["target_technologies"])
tech_overlap = len(techs & target_techs)
tech_score = min(15, tech_overlap * 5)
score += tech_score
breakdown["technology_fit"] = {"score": tech_score, "max": 15, "detail": list(techs & target_techs)}
# Géo (10 pts)
country = company_data.get("location", {}).get("country", "")
if country in icp_criteria["target_geos"]:
score += 10
breakdown["geography"] = {"score": 10, "max": 10, "detail": country}
else:
breakdown["geography"] = {"score": 0, "max": 10, "detail": country}
tier = "A+" if score >= 85 else "A" if score >= 70 else "B" if score >= 50 else "C" if score >= 30 else "D"
return json.dumps({
"total_score": round(score),
"tier": tier,
"breakdown": breakdown,
"recommendation": "HIGH_PRIORITY" if tier in ["A+", "A"] else
"STANDARD" if tier == "B" else "LOW_PRIORITY"
}, indent=2)
@mcp.tool()
async def generate_outreach(
prospect_name: str,
prospect_title: str,
company_name: str,
company_data: dict,
value_proposition: str,
tone: str = "professional",
language: str = "fr"
) -> str:
"""Génère un email de prospection personnalisé.
Le contenu est un BROUILLON à valider par le commercial."""
# La génération est faite par Claude lui-même
return json.dumps({
"status": "DRAFT",
"instructions": f"Générez un email pour {prospect_name} ({prospect_title}) "
f"chez {company_name}. Ton: {tone}, Langue: {language}. "
f"Personnalisez avec les données enrichies et la value prop.",
"template_sections": [
"subject_line",
"opening_hook",
"pain_point",
"value_proposition",
"social_proof",
"cta",
"ps_line"
],
"validation_required": True
})
@mcp.tool()
async def sync_to_hubspot(
contact_data: dict,
deal_data: dict = None
) -> str:
"""Synchronise un contact (et optionnellement un deal) dans HubSpot.
Crée ou met à jour le contact, crée un deal si fourni."""
async with httpx.AsyncClient() as client:
# Créer/mettre à jour le contact
contact_resp = await client.post(
f"{HUBSPOT_API}/objects/contacts",
headers={
"Authorization": f"Bearer {HUBSPOT_TOKEN}",
"Content-Type": "application/json"
},
json={
"properties": {
"email": contact_data["email"],
"firstname": contact_data.get("first_name", ""),
"lastname": contact_data.get("last_name", ""),
"jobtitle": contact_data.get("title", ""),
"company": contact_data.get("company", ""),
"lead_source": "AI_PROSPECTION",
"icp_score": str(contact_data.get("score", 0))
}
}
)
contact_id = contact_resp.json().get("id")
deal_id = None
if deal_data:
deal_resp = await client.post(
f"{HUBSPOT_API}/objects/deals",
headers={
"Authorization": f"Bearer {HUBSPOT_TOKEN}",
"Content-Type": "application/json"
},
json={
"properties": {
"dealname": deal_data.get("name", f"Prospect - {contact_data.get('company', 'N/A')}"),
"pipeline": "default",
"dealstage": "qualifiedtobuy",
"amount": str(deal_data.get("amount", 0))
}
}
)
deal_id = deal_resp.json().get("id")
return json.dumps({
"contact_id": contact_id,
"deal_id": deal_id,
"status": "SYNCED",
"hubspot_url": f"https://app.hubspot.com/contacts/{contact_id}"
}, indent=2)
if __name__ == "__main__":
mcp.run(transport="stdio")#!/bin/bash
# Pipeline de prospection autonome quotidien
# Exécuté chaque matin à 7h30
DATE=$(date +%Y-%m-%d)
echo "=== PROSPECTION PIPELINE ${DATE} ==="
# Étape 1 : Identifier 20 leads potentiels
claude -p "Recherche 20 entreprises correspondant à notre ICP :
- SaaS B2B, 50-500 employés, Europe ou Amérique du Nord
- Utilise : Python, AWS, ou Kubernetes
- Série A minimum
- Qui n'est PAS déjà dans notre CRM HubSpot
Utilise enrich_company() pour chaque domaine trouvé.
Score chaque lead avec score_lead().
Retourne les 10 meilleurs (tier A ou A+)." \
--output-format json > /tmp/leads-${DATE}.json
# Étape 2 : Trouver les décideurs
claude -p "Pour chaque lead dans /tmp/leads-${DATE}.json,
trouve les décideurs clés avec find_decision_makers().
Cible : CTO, VP Engineering, Head of Platform.
Synthétise dans un fichier CSV." \
--output-format json > /tmp/contacts-${DATE}.json
# Étape 3 : Générer les emails (DRAFT)
claude -p "Pour les 5 contacts les mieux scorés de /tmp/contacts-${DATE}.json :
1. Génère un email de prospection personnalisé en français
2. Référence une technologie ou un événement récent de l'entreprise
3. Propose notre value prop : 'automatisation IA des processus métier'
4. Marque tous les emails comme DRAFT pour validation commerciale
5. Sync les contacts dans HubSpot avec sync_to_hubspot()" \
--output-format json > /tmp/outreach-${DATE}.json
echo "=== ${DATE} : Prospection terminée ==="
echo "Leads identifiés : $(jq '.leads | length' /tmp/leads-${DATE}.json)"
echo "Emails DRAFT générés : $(jq '.drafts | length' /tmp/outreach-${DATE}.json)"Le reporting consume un temps disproportionné dans toute organisation. Un analyste passe en moyenne 40% de son temps à collecter des données, 30% à les formater, et seulement 30% à les analyser. Claude Code inverse cette équation : 0% collecte, 0% formatage, 100% analyse et insight.
| Type de rapport | Fréquence | Destinataire | Contenu clé |
|---|---|---|---|
| Flash Report | Quotidien (8h) | CEO / COO | KPIs critiques, alertes, actions du jour |
| Sales Report | Hebdomadaire | VP Sales | Pipeline, conversion, forecast, top deals |
| Financial Report | Mensuel | CFO / Board | P&L, bilan, cash flow, variance analysis |
| Marketing Report | Hebdomadaire | CMO | CAC, LTV, attribution, ROI campagnes |
| Operations Report | Quotidien | VP Ops | SLA, uptime, incidents, throughput |
| Board Report | Trimestriel | Conseil d'admin | Executive summary, stratégie, financials |
Data Warehouse Semantic Layer Claude Code Delivery
┌──────────┐ ┌──────────┐ ┌──────────────┐ ┌──────────┐
│ Snowflake│──────▶│ dbt SL │──────▶│ Query data │────▶│ Slack │
│ BigQuery │ │ Cube.js │ │ Detect anom. │ │ Email │
│ Postgres │ │ │ │ Gen narrative │ │ Notion │
└──────────┘ └──────────┘ │ Create charts│ │ PDF │
│ Format report│ │ Dashboard│
└──────────────┘ └──────────┘
#!/bin/bash
# Génération automatique du rapport hebdomadaire
# Exécuté chaque lundi à 7h00
WEEK=$(date -d "last monday" +%Y-W%V)
REPORT_DIR="/reports/weekly/${WEEK}"
mkdir -p "${REPORT_DIR}"
# Phase 1 : Collecte de données via le Semantic Layer
claude -p "Collecte les données de la semaine passée via le Semantic Layer :
1. **Revenue** : query_metric(['mrr', 'new_mrr', 'churned_mrr', 'net_mrr_retention'],
time_grain='week', time_range='last_2_weeks', group_by=['customer_segment'])
2. **Sales** : query_metric(['pipeline_value', 'deals_won', 'win_rate', 'avg_deal_size'],
time_grain='week', time_range='last_2_weeks', group_by=['sales_rep'])
3. **Marketing** : query_metric(['cac', 'ltv', 'signups', 'trial_to_paid_rate'],
time_grain='week', time_range='last_2_weeks', group_by=['channel'])
4. **Support** : query_metric(['tickets_created', 'resolution_time', 'csat_score'],
time_grain='week', time_range='last_2_weeks')
Compare chaque métrique avec la semaine précédente.
Sauvegarde les données brutes dans ${REPORT_DIR}/data.json" \
--output-format json > "${REPORT_DIR}/data.json"
# Phase 2 : Détection d'anomalies
claude -p "Analyse les données de ${REPORT_DIR}/data.json.
Identifie toute métrique qui :
- Varie de plus de 15% vs semaine précédente
- Est en dessous du target/budget
- Présente une tendance négative sur 3+ semaines
- Atteint un nouveau record (positif ou négatif)
Pour chaque anomalie, propose une explication et une action.
Format JSON." \
--output-format json > "${REPORT_DIR}/anomalies.json"
# Phase 3 : Génération du rapport narratif
claude -p "Génère le rapport hebdomadaire ${WEEK} en HTML.
Données : ${REPORT_DIR}/data.json
Anomalies : ${REPORT_DIR}/anomalies.json
Structure :
1. Executive Summary (5 bullet points max)
2. KPI Dashboard (tableau avec semaine N vs N-1 vs target)
3. Revenue & MRR (graphique + narrative)
4. Sales Pipeline (graphique + narrative)
5. Marketing Performance (graphique + narrative)
6. Support Quality (graphique + narrative)
7. Anomalies & Alertes
8. Actions recommandées
9. Outlook semaine prochaine
Style : professionnel, concis, data-driven.
Utilise ECharts pour les graphiques.
Langue : français." > "${REPORT_DIR}/report.html"
# Phase 4 : Distribution
claude -p "Distribue le rapport ${REPORT_DIR}/report.html :
1. Envoie par email au comité de direction (cfo@, ceo@, vp-sales@)
2. Poste un résumé dans #weekly-report sur Slack
3. Archive dans Notion (page 'Weekly Reports / ${WEEK}')
Sujet email : 'Rapport Hebdomadaire ${WEEK} - [MRR] [Anomalies]'"
echo "=== Rapport ${WEEK} généré et distribué ===""""
Module de détection d'anomalies pour le reporting automatisé.
Utilise des méthodes statistiques simples + Claude pour l'interprétation.
"""
import numpy as np
from dataclasses import dataclass
@dataclass
class Anomaly:
metric: str
current_value: float
expected_value: float
deviation_percent: float
severity: str # LOW, MEDIUM, HIGH, CRITICAL
direction: str # UP, DOWN
context: str
def detect_anomalies(
metric_name: str,
values: list[float],
labels: list[str],
threshold_sigma: float = 2.0,
min_deviation_pct: float = 15.0
) -> list[Anomaly]:
"""Détecte les anomalies dans une série temporelle de métriques.
Méthode : Z-score + seuil de déviation relative.
Une anomalie est signalée si :
1. Le Z-score dépasse threshold_sigma (défaut: 2.0)
2. ET la déviation relative dépasse min_deviation_pct (défaut: 15%)
"""
if len(values) < 4:
return []
arr = np.array(values)
mean = np.mean(arr[:-1]) # Moyenne sans la dernière valeur
std = np.std(arr[:-1])
current = arr[-1]
anomalies = []
if std > 0:
z_score = (current - mean) / std
deviation_pct = abs(current - mean) / mean * 100 if mean != 0 else 0
if abs(z_score) > threshold_sigma and deviation_pct > min_deviation_pct:
severity = ("CRITICAL" if abs(z_score) > 3.5 else
"HIGH" if abs(z_score) > 3.0 else
"MEDIUM" if abs(z_score) > 2.5 else "LOW")
anomalies.append(Anomaly(
metric=metric_name,
current_value=current,
expected_value=round(mean, 2),
deviation_percent=round(deviation_pct, 1),
severity=severity,
direction="UP" if current > mean else "DOWN",
context=f"Z-score: {z_score:.2f}, {len(values)-1} périodes analysées"
))
# Détection de tendance (3+ périodes consécutives en baisse/hausse)
if len(values) >= 4:
recent = values[-4:]
diffs = [recent[i+1] - recent[i] for i in range(3)]
if all(d < 0 for d in diffs):
anomalies.append(Anomaly(
metric=metric_name,
current_value=current,
expected_value=values[-4],
deviation_percent=round(abs(current - values[-4]) / values[-4] * 100, 1) if values[-4] != 0 else 0,
severity="MEDIUM",
direction="DOWN",
context="Tendance baissiere sur 4 periodes consecutives"
))
elif all(d > 0 for d in diffs):
anomalies.append(Anomaly(
metric=metric_name,
current_value=current,
expected_value=values[-4],
deviation_percent=round(abs(current - values[-4]) / values[-4] * 100, 1) if values[-4] != 0 else 0,
severity="LOW",
direction="UP",
context="Tendance haussiere sur 4 periodes consecutives"
))
return anomaliesL'intégration du Semantic Layer avec Claude permet le Natural Language BI : poser des questions métier en français et obtenir des réponses avec données, graphiques et recommandations.
# Question simple
$ claude -p "Quel est notre MRR ce mois ?"
→ query_metric(["mrr"], time_range="this_month")
→ "Votre MRR est de 847 320€, en hausse de 4.2% vs le mois dernier."
# Question comparative
$ claude -p "Pourquoi notre churn a augmenté de 12% ce mois ?"
→ query_metric(["churn_rate"], group_by=["customer_segment", "plan_type"], time_range="last_2_months")
→ compare_periods("churn_rate", "this_month", "last_month", ["customer_segment"])
→ "Le churn a augmenté principalement sur le segment startup (+23%)
et le plan Free (+31%). Cause probable : expiration des périodes d'essai
de la campagne Q3. Recommandation : offre de rétention ciblée."
# Question prédictive
$ claude -p "Va-t-on atteindre notre target de 1M MRR d'ici fin Q1 ?"
→ query_metric(["mrr"], time_grain="month", time_range="last_6_months")
→ "Projection linéaire : 962K€ fin Q1 (-3.8% vs target).
Pour atteindre 1M, il faut ajouter 38K MRR net/mois vs 29K actuel (+31%).
Leviers : réduire le churn de 2pp ou augmenter les upgrades de 15%."L'orchestration est le liant qui connecte toutes les automatisations. Un processus métier n'est jamais un outil isolé — c'est une chaîne de décisions et d'actions impliquant plusieurs systèmes, plusieurs personnes, et de multiples points de contrôle. Claude Code, couplé à un moteur de workflow, peut orchestrer ces chaînes de bout en bout.
| Moteur | Type | Intégration Claude | Cas d'usage |
|---|---|---|---|
| n8n | Low-code, self-hosted | HTTP nodes + code nodes | Workflows simples, intégrations SaaS |
| Temporal | Code-first, durable | Activities + Workers | Workflows complexes, fiabilité critique |
| Inngest | Event-driven, serverless | Fonctions step | Microservices, event-driven |
| Trigger.dev | Background jobs, serverless | SDK TypeScript natif | Tâches asynchrones, intégrations |
| Prefect | Data orchestration, Python | Task decorators | Data pipelines, ML workflows |
Événement Claude Décide Action Vérification
┌──────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────┐
│ Trigger │─────▶│ Analyse le │───▶│ Exécute via │───▶│ Vérifie │
│ │ │ contexte │ │ MCP server │ │ résultat │
│ Webhook │ │ Décide quoi │ │ ou Bash │ │ │
│ Cron │ │ faire │ │ │ │ OK ? │
│ Email │ │ Route vers │ │ Ou route │ │ ──▶ Log │
│ Slack │ │ le bon skill │ │ vers humain │ │ NOK? │
│ API call │ │ │ │ │ │ ──▶ Retry│
└──────────┘ └──────────────┘ └──────────────┘ └──────────┘
"""
Workflow Temporal : Onboarding automatisé d'un nouvel employé.
10 étapes orchestrées avec Claude Code comme moteur décisionnel.
"""
from datetime import timedelta
from temporalio import workflow, activity
from temporalio.common import RetryPolicy
import subprocess
import json
# Politique de retry standard
RETRY = RetryPolicy(
initial_interval=timedelta(seconds=10),
maximum_interval=timedelta(minutes=5),
maximum_attempts=3,
non_retryable_error_types=["HumanApprovalRequired"]
)
@activity.defn
async def create_accounts(employee: dict) -> dict:
"""Étape 1 : Créer les comptes (Google Workspace, Slack, GitHub)."""
result = subprocess.run(
["claude", "-p", json.dumps({
"task": "Crée les comptes pour le nouvel employé",
"employee": employee,
"accounts": ["google_workspace", "slack", "github", "notion"],
"instructions": "Utilise les MCP servers pour créer chaque compte. "
"Génère un mot de passe temporaire pour chaque service. "
"Ajoute aux bons groupes/channels selon le département."
}), "--output-format", "json"],
capture_output=True, text=True
)
return json.loads(result.stdout)
@activity.defn
async def setup_hardware(employee: dict) -> dict:
"""Étape 2 : Commander et configurer le matériel."""
result = subprocess.run(
["claude", "-p", f"""Configure le matériel pour {employee['name']} :
- Rôle : {employee['role']}
- Département : {employee['department']}
Détermine le package matériel standard :
- Développeur : MacBook Pro 16" M4, écran 27", clavier/souris
- Commercial : MacBook Air 15" M3, écran 24"
- Design : MacBook Pro 16" M4 Max, écran 32" 4K, tablette
Crée le ticket IT dans Jira avec les spécifications.""",
"--output-format", "json"],
capture_output=True, text=True
)
return json.loads(result.stdout)
@activity.defn
async def generate_onboarding_docs(employee: dict) -> dict:
"""Étape 3 : Générer les documents d'onboarding personnalisés."""
result = subprocess.run(
["claude", "-p", f"""Génère les documents d'onboarding pour {employee['name']} ({employee['role']}) :
1. Welcome letter personnalisée (mentionner l'équipe, le manager, les projets)
2. Guide de démarrage spécifique au rôle ({employee['department']})
3. Liste des formations obligatoires (sécurité, RGPD, culture d'entreprise)
4. Planning de la première semaine (meetings, buddy, déjeuners)
5. FAQ personnalisée selon le département
Langue : français. Ton : chaleureux et professionnel.
Sauvegarde dans Notion sous 'Onboarding / {employee['name']}'.""",
"--output-format", "json"],
capture_output=True, text=True
)
return json.loads(result.stdout)
@activity.defn
async def assign_buddy(employee: dict) -> dict:
"""Étape 4 : Assigner un buddy en fonction du département et de la dispo."""
result = subprocess.run(
["claude", "-p", f"""Assigne un buddy pour {employee['name']} ({employee['department']}) :
Critères de sélection :
- Même département ou département proche
- Ancienneté > 6 mois
- N'est pas déjà buddy d'un autre nouvel arrivant
- Disponible les 2 prochaines semaines
Interroge le SIRH via MCP pour trouver les candidats.
Envoie une invitation Slack au buddy sélectionné.""",
"--output-format", "json"],
capture_output=True, text=True
)
return json.loads(result.stdout)
@activity.defn
async def schedule_meetings(employee: dict, buddy: dict) -> dict:
"""Étape 5 : Planifier les meetings de la première semaine."""
result = subprocess.run(
["claude", "-p", f"""Planifie les meetings d'onboarding pour {employee['name']} :
Semaine 1 :
- Lundi 9h : Welcome meeting avec le manager ({employee['manager']})
- Lundi 11h : Setup technique avec IT
- Lundi 12h : Déjeuner avec le buddy ({buddy['name']})
- Mardi 10h : Présentation produit
- Mercredi 14h : Culture & valeurs
- Jeudi 10h : Formation sécurité / RGPD
- Vendredi 16h : 1:1 fin de première semaine avec manager
Utilise Google Calendar API pour créer les événements.
Vérifie la disponibilité de chaque participant.""",
"--output-format", "json"],
capture_output=True, text=True
)
return json.loads(result.stdout)
@activity.defn
async def notify_team(employee: dict) -> dict:
"""Étape 6 : Notifier l'équipe de l'arrivée du nouvel employé."""
result = subprocess.run(
["claude", "-p", f"""Annonce l'arrivée de {employee['name']} :
1. Message Slack dans #{employee['department']} :
- Présentation courte (rôle, background, fun facts)
- Photo si disponible
- Date de début
2. Email au manager avec le résumé de l'onboarding :
- Comptes créés
- Matériel commandé
- Buddy assigné
- Planning semaine 1""",
"--output-format", "json"],
capture_output=True, text=True
)
return json.loads(result.stdout)
@workflow.defn
class EmployeeOnboardingWorkflow:
"""Workflow principal d'onboarding - 10 étapes."""
@workflow.run
async def run(self, employee: dict) -> dict:
results = {}
# Étapes parallélisables (1-3)
accounts, hardware, docs = await workflow.execute_activities(
[create_accounts, setup_hardware, generate_onboarding_docs],
args=[[employee], [employee], [employee]],
start_to_close_timeout=timedelta(minutes=10),
retry_policy=RETRY
)
results["accounts"] = accounts
results["hardware"] = hardware
results["docs"] = docs
# Étape 4 : Buddy (dépend des comptes)
buddy = await workflow.execute_activity(
assign_buddy,
args=[employee],
start_to_close_timeout=timedelta(minutes=5),
retry_policy=RETRY
)
results["buddy"] = buddy
# Étape 5 : Meetings (dépend du buddy)
meetings = await workflow.execute_activity(
schedule_meetings,
args=[employee, buddy],
start_to_close_timeout=timedelta(minutes=5),
retry_policy=RETRY
)
results["meetings"] = meetings
# Étape 6 : Notification équipe
await workflow.execute_activity(
notify_team,
args=[employee],
start_to_close_timeout=timedelta(minutes=5),
retry_policy=RETRY
)
return results#!/bin/bash
# Workflow de réponse à incident déclenché par PagerDuty/Grafana
# Args : $1 = severity, $2 = service, $3 = description
SEVERITY=$1
SERVICE=$2
DESCRIPTION=$3
INCIDENT_ID="INC-$(date +%Y%m%d%H%M%S)"
echo "=== INCIDENT ${INCIDENT_ID} | ${SEVERITY} | ${SERVICE} ==="
# Étape 1 : Analyse initiale
claude -p "Incident détecté :
- ID: ${INCIDENT_ID}
- Sévérité: ${SEVERITY}
- Service: ${SERVICE}
- Description: ${DESCRIPTION}
1. Analyse les logs récents du service (via MCP logging)
2. Vérifie le statut des dépendances
3. Identifie la cause probable
4. Propose un plan d'action immédiat
5. Détermine si une escalation humaine est nécessaire" \
--output-format json > /tmp/incident-${INCIDENT_ID}-analysis.json
# Étape 2 : Communication
claude -p "Envoie les notifications pour l'incident ${INCIDENT_ID} :
1. Alerte Slack #incidents avec le résumé
2. Si severity=P1 : appeler le on-call via PagerDuty
3. Créer le ticket Jira dans le projet OPS
4. Mettre à jour la status page (si impact client)"
# Étape 3 : Action corrective (si automatisée)
if [ "${SEVERITY}" != "P1" ]; then
claude -p "Exécute les actions correctives automatisées pour ${INCIDENT_ID} :
- Restart du service si health check échoue
- Scale-up si CPU/Memory > 90%
- Rollback si deploy récent (< 30min)
- Clear cache si erreurs 5xx"
fi
echo "=== INCIDENT ${INCIDENT_ID} TRAITÉ ==="Les pipelines de données traditionnels sont fragiles : un changement de schéma casse le pipeline, une anomalie dans les données source passe inaperçue, et le debugging nécessite des heures d'investigation. Claude Code transforme les pipelines passifs en pipelines intelligents capables de détecter, diagnostiquer et réparer les problèmes automatiquement.
| Aspect | ETL Traditionnel | ETL Augmenté par Claude |
|---|---|---|
| Mapping de schéma | Manuel, codé en dur | Détection automatique, suggére les mappings |
| Qualité des données | Règles statiques (NOT NULL, FK) | Anomalies statistiques + sémantiques |
| Gestion d'erreurs | Retry ou fail | Diagnostic, correction automatique, retry intelligent |
| Changement de source | Pipeline cassé, intervention manuelle | Détection du changement, adaptation suggérée |
| Documentation | Souvent absente ou obsolète | Générée automatiquement, toujours à jour |
| PII Detection | Règles regex basiques | Compréhension contextuelle (noms, adresses, etc.) |
-- Macro dbt qui génère des tests de qualité de données via Claude
-- Appelée dans le pipeline après chaque transformation
{% macro claude_anomaly_check(model_name, column_name, lookback_days=30) %}
WITH stats AS (
SELECT
AVG({{ column_name }}) as avg_val,
STDDEV({{ column_name }}) as std_val,
MIN({{ column_name }}) as min_val,
MAX({{ column_name }}) as max_val,
COUNT(*) as row_count,
COUNT(DISTINCT {{ column_name }}) as distinct_count,
SUM(CASE WHEN {{ column_name }} IS NULL THEN 1 ELSE 0 END) as null_count
FROM {{ ref(model_name) }}
WHERE created_at >= CURRENT_DATE - INTERVAL '{{ lookback_days }} days'
),
latest AS (
SELECT
AVG({{ column_name }}) as latest_avg,
COUNT(*) as latest_count
FROM {{ ref(model_name) }}
WHERE created_at >= CURRENT_DATE - INTERVAL '1 day'
)
SELECT
'{{ model_name }}.{{ column_name }}' as check_target,
CASE
WHEN latest.latest_avg > stats.avg_val + 3 * stats.std_val THEN 'ANOMALY_HIGH'
WHEN latest.latest_avg < stats.avg_val - 3 * stats.std_val THEN 'ANOMALY_LOW'
WHEN latest.latest_count < stats.row_count / {{ lookback_days }} * 0.5 THEN 'LOW_VOLUME'
WHEN stats.null_count > stats.row_count * 0.1 THEN 'HIGH_NULLS'
ELSE 'OK'
END as status,
stats.*,
latest.*
FROM stats, latest
WHERE CASE
WHEN latest.latest_avg > stats.avg_val + 3 * stats.std_val THEN TRUE
WHEN latest.latest_avg < stats.avg_val - 3 * stats.std_val THEN TRUE
WHEN latest.latest_count < stats.row_count / {{ lookback_days }} * 0.5 THEN TRUE
WHEN stats.null_count > stats.row_count * 0.1 THEN TRUE
ELSE FALSE
END
{% endmacro %}"""
Pipeline de données auto-réparant avec Claude Code.
Détecte les problèmes, diagnostique et propose des corrections.
"""
import subprocess
import json
import sys
def run_dbt_and_heal():
"""Exécute dbt run et répare automatiquement les erreurs courantes."""
# Exécuter dbt
result = subprocess.run(
["dbt", "run", "--target", "prod"],
capture_output=True, text=True
)
if result.returncode == 0:
print("dbt run OK - aucune erreur")
# Exécuter les tests de qualité
test_result = subprocess.run(
["dbt", "test", "--target", "prod"],
capture_output=True, text=True
)
if test_result.returncode != 0:
diagnose_test_failures(test_result.stderr)
return
# Erreur : demander à Claude de diagnostiquer
error_output = result.stderr[-3000:] # Limiter la taille
diagnosis = subprocess.run(
["claude", "-p", f"""Le pipeline dbt a échoué. Voici l'erreur :
{error_output}
Diagnostique le problème et propose une solution :
1. Type d'erreur (schema change, permission, timeout, data issue)
2. Modèle concerné
3. Cause probable
4. Action corrective
5. Est-ce que la correction peut être appliquée automatiquement ? (oui/non)
Si oui, génère le correctif (SQL ou YAML).
Format : JSON avec les champs ci-dessus.""",
"--output-format", "json"],
capture_output=True, text=True
)
diag = json.loads(diagnosis.stdout)
if diag.get("auto_fixable"):
print(f"Correction automatique : {diag['action']}")
# Appliquer le correctif
apply_fix(diag)
# Re-run dbt
retry = subprocess.run(
["dbt", "run", "--select", diag["model"], "--target", "prod"],
capture_output=True, text=True
)
if retry.returncode == 0:
print("Pipeline réparé automatiquement")
notify_slack(f"Pipeline réparé auto : {diag['action']}")
else:
escalate(diag, retry.stderr)
else:
escalate(diag, error_output)
def diagnose_test_failures(error_output):
"""Utilise Claude pour analyser les échecs de tests dbt."""
subprocess.run(
["claude", "-p", f"""Les tests dbt ont échoué :
{error_output}
Pour chaque test en échec :
1. Identifie le type d'anomalie (nulls, duplicates, referential integrity, freshness)
2. Quantifie l'impact (combien de rows affectées)
3. Propose une explication
4. Recommande : BLOCK (arrêter le pipeline) ou WARN (continuer avec alerte)"""],
capture_output=True, text=True
)
def apply_fix(diagnosis):
"""Applique une correction automatique."""
if diagnosis["type"] == "schema_change":
# Ajouter la colonne manquante au modèle
subprocess.run(
["claude", "-p", f"Mets à jour le modèle dbt {diagnosis['model']} "
f"pour gérer le changement de schéma : {diagnosis['fix']}"],
capture_output=True, text=True
)
elif diagnosis["type"] == "timeout":
# Optimiser la requête
subprocess.run(
["claude", "-p", f"Optimise la requête SQL du modèle {diagnosis['model']}. "
f"Problème : timeout. Suggère des index ou une refactorisation."],
capture_output=True, text=True
)
def escalate(diagnosis, error):
"""Escalade à l'humain quand la réparation automatique n'est pas possible."""
notify_slack(
f"PIPELINE FAILURE - Escalation requise\n"
f"Modèle : {diagnosis.get('model', 'unknown')}\n"
f"Type : {diagnosis.get('type', 'unknown')}\n"
f"Diagnostic : {diagnosis.get('cause', 'unknown')}"
)
def notify_slack(message):
"""Envoie une notification Slack."""
subprocess.run(
["curl", "-X", "POST", "-H", "Content-type: application/json",
"--data", json.dumps({"text": message}),
"https://hooks.slack.com/services/T.../B.../xxx"],
capture_output=True
)
if __name__ == "__main__":
run_dbt_and_heal()"""
Détection de PII (Personally Identifiable Information) via Claude.
Avantage vs regex : compréhension contextuelle (un numéro de téléphone
dans un champ 'emergency_contact' est du PII, dans 'store_phone' non).
"""
from mcp.server.fastmcp import FastMCP
import json
mcp = FastMCP("data-catalog")
PII_CATEGORIES = {
"DIRECT_ID": "Identifiants directs (nom, email, SSN, passeport)",
"QUASI_ID": "Quasi-identifiants (date naissance, code postal, genre)",
"SENSITIVE": "Données sensibles (santé, religion, opinions politiques)",
"FINANCIAL": "Données financières (IBAN, carte bancaire, salaire)",
"NONE": "Aucune donnée personnelle détectée"
}
@mcp.tool()
async def classify_columns(
table_name: str,
columns: list[dict],
sample_rows: list[dict] = None
) -> str:
"""Classifie les colonnes d'une table selon leur sensibilité PII.
Args:
table_name: Nom de la table
columns: [{"name": "col_name", "type": "varchar", "sample": ["val1", "val2"]}]
sample_rows: Échantillon de lignes pour analyse contextuelle
"""
classifications = []
for col in columns:
# L'analyse est déléguée à Claude qui comprend le contexte
classifications.append({
"column": col["name"],
"type": col["type"],
"instructions": f"Classifie '{col['name']}' (type: {col['type']}) "
f"avec échantillon : {col.get('sample', [])}. "
f"Catégories : {list(PII_CATEGORIES.keys())}"
})
return json.dumps({
"table": table_name,
"classifications": classifications,
"categories_reference": PII_CATEGORIES
}, indent=2)
@mcp.tool()
async def generate_data_catalog_entry(
table_name: str,
schema: list[dict],
source: str,
owner: str
) -> str:
"""Génère automatiquement une entrée de catalogue de données.
Inclut : description, lineage, PII classification, SLA, owner."""
return json.dumps({
"instructions": f"Génère une entrée de catalogue pour '{table_name}' : "
f"description métier, classification PII de chaque colonne, "
f"SLA de fraîcheur, owner ({owner}), source ({source}), "
f"modèles dbt qui en dépendent (lineage downstream)."
})
if __name__ == "__main__":
mcp.run(transport="stdio")Le support client est le domaine où l'IA d'entreprise a le plus d'impact immédiat. Klarna a réduit son équipe support de 700 personnes en déployant un assistant IA qui gère 2/3 des interactions client. Résultat : temps de résolution divisé par 5, satisfaction client stable. Voici comment construire un système équivalent avec Claude Code.
Canal d'entrée Triage IA Résolution Escalation
┌──────────┐ ┌──────────┐ ┌──────────────┐ ┌──────────┐
│ Email │─────▶│ Claude │─────▶│ L1 : IA auto │ │ L2 : IA │
│ Chat │ │ classifie│ │ 70% tickets │ │ + humain │
│ Téléphone │ │ priorise │ │ │ │ 25% │
│ Slack │ │ route │ │ FAQ, password │ │ │
│ Twitter │ │ │ │ status, refund│ │ Complex │
└──────────┘ └──────────┘ └──────────────┘ │ Technic. │
│ └──────────┘
│ │
│ ┌──────────────┐ ┌──────────┐
└────────────▶│ L3 : Humain │ │ Feedback │
│ expert │ │ Loop │
│ 5% tickets │ │ Train IA │
│ │ │ │
│ Juridique │ └──────────┘
│ Sécurité │
└──────────────┘
"""
MCP Server : Customer Support Engine
Triage, réponse, escalation, knowledge base.
"""
import json
import httpx
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("support")
ZENDESK_URL = "https://company.zendesk.com/api/v2"
ZENDESK_TOKEN = ""
ZENDESK_EMAIL = "[email protected]"
auth = (f"{ZENDESK_EMAIL}/token", ZENDESK_TOKEN)
@mcp.tool()
async def get_ticket(ticket_id: int) -> str:
"""Récupère un ticket Zendesk avec son historique complet."""
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{ZENDESK_URL}/tickets/{ticket_id}.json?include=users,comment_count",
auth=auth
)
ticket = resp.json()["ticket"]
# Récupérer les commentaires
comments_resp = await client.get(
f"{ZENDESK_URL}/tickets/{ticket_id}/comments.json",
auth=auth
)
comments = comments_resp.json()["comments"]
return json.dumps({
"id": ticket["id"],
"subject": ticket["subject"],
"status": ticket["status"],
"priority": ticket["priority"],
"requester_id": ticket["requester_id"],
"created_at": ticket["created_at"],
"tags": ticket["tags"],
"comments": [{
"author": c.get("author_id"),
"body": c["body"][:500],
"created_at": c["created_at"],
"public": c["public"]
} for c in comments]
}, indent=2)
@mcp.tool()
async def classify_ticket(
subject: str,
body: str
) -> str:
"""Classifie un ticket support :
- Catégorie (billing, technical, feature_request, bug, account)
- Priorité (low, normal, high, urgent)
- Sentiment (positive, neutral, negative, angry)
- Complexité (L1_auto, L2_assisted, L3_expert)"""
return json.dumps({
"instructions": "Classifiez ce ticket :\n"
f"Sujet : {subject}\n"
f"Corps : {body[:1000]}\n\n"
"Retournez : category, priority, sentiment, complexity, "
"suggested_tags, estimated_resolution_time"
})
@mcp.tool()
async def search_knowledge_base(query: str, limit: int = 5) -> str:
"""Recherche dans la knowledge base Zendesk pour trouver des réponses."""
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{ZENDESK_URL}/help_center/articles/search.json",
auth=auth,
params={"query": query, "per_page": limit}
)
articles = resp.json().get("results", [])
return json.dumps([{
"id": a["id"],
"title": a["title"],
"snippet": a.get("snippet", "")[:300],
"url": a["html_url"],
"vote_sum": a.get("vote_sum", 0)
} for a in articles], indent=2)
@mcp.tool()
async def draft_response(
ticket_id: int,
category: str,
customer_name: str,
issue_summary: str,
kb_articles: list[dict] = None,
tone: str = "empathetic_professional"
) -> str:
"""Génère un BROUILLON de réponse pour un ticket support.
Le brouillon est enregistré comme note interne (non envoyé)."""
return json.dumps({
"instructions": f"Générez une réponse support pour le ticket #{ticket_id}.\n"
f"Client : {customer_name}\n"
f"Problème : {issue_summary}\n"
f"Catégorie : {category}\n"
f"Ton : {tone}\n"
f"Articles KB : {kb_articles}\n\n"
"Structure : Salutation personnalisée > Reconnaissance du problème > "
"Solution claire et étapée > Lien KB si pertinent > Offre d'aide supplémentaire > Signature",
"save_as": "internal_note",
"requires_approval": True
})
@mcp.tool()
async def update_ticket(
ticket_id: int,
status: str = None,
priority: str = None,
tags: list[str] = None,
assignee_id: int = None,
comment: str = None,
public: bool = False
) -> str:
"""Met à jour un ticket Zendesk (statut, priorité, tags, assignation)."""
update = {"ticket": {}}
if status: update["ticket"]["status"] = status
if priority: update["ticket"]["priority"] = priority
if tags: update["ticket"]["tags"] = tags
if assignee_id: update["ticket"]["assignee_id"] = assignee_id
if comment:
update["ticket"]["comment"] = {"body": comment, "public": public}
async with httpx.AsyncClient() as client:
resp = await client.put(
f"{ZENDESK_URL}/tickets/{ticket_id}.json",
auth=auth,
json=update
)
return json.dumps({"status": "updated", "ticket_id": ticket_id})
@mcp.tool()
async def get_customer_history(requester_id: int) -> str:
"""Récupère l'historique d'un client : tickets précédents, CSAT, plan."""
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{ZENDESK_URL}/users/{requester_id}/tickets/requested.json",
auth=auth,
params={"sort_by": "created_at", "sort_order": "desc", "per_page": 10}
)
tickets = resp.json().get("tickets", [])
return json.dumps({
"total_tickets": len(tickets),
"recent_tickets": [{
"id": t["id"],
"subject": t["subject"],
"status": t["status"],
"created_at": t["created_at"],
"satisfaction_rating": t.get("satisfaction_rating", {}).get("score", "N/A")
} for t in tickets[:10]],
"avg_resolution_time": "computed_by_claude",
"churn_risk": "computed_by_claude"
}, indent=2)
if __name__ == "__main__":
mcp.run(transport="stdio")# /handle-ticket
Traitement automatique d'un ticket support L1.
## Workflow
1. **get_ticket()** : Récupérer le ticket complet
2. **classify_ticket()** : Catégoriser, prioriser, détecter le sentiment
3. **get_customer_history()** : Contexte client (tickets précédents, plan, risque churn)
4. **search_knowledge_base()** : Trouver les articles pertinents
5. **Décision** :
- Si L1_auto ET sentiment != angry : générer réponse automatique
- Si L2_assisted : générer brouillon + assigner à agent humain
- Si L3_expert : escalader directement sans réponse auto
- Si client VIP (plan Enterprise) : escalader systématiquement
6. **draft_response()** : Générer la réponse (brouillon ou auto)
7. **update_ticket()** : Mettre à jour statut, tags, assignation
## Règles de sécurité
- NE JAMAIS promettre un remboursement > 100€ sans validation humaine
- NE JAMAIS partager des données personnelles d'un autre client
- NE JAMAIS mentionner des fonctionnalités futures non annoncées
- Toujours proposer l'option "parler à un humain"
- Réponses auto UNIQUEMENT pour les catégories validées (billing simple, FAQ, status)Le département juridique est souvent le goulot d'étranglement de l'entreprise : chaque contrat, chaque partenariat, chaque nouvelle fonctionnalité doit passer par la revue juridique. Claude Code peut accélérer ce processus de 10x en automatisant la revue initiale, l'extraction de clauses, et la veille réglementaire.
| Processus | Automatisation | Gain de temps | Risque |
|---|---|---|---|
| Revue de contrat | Extraction clauses, scoring risque, benchmark | -75% | Moyen (validation requise) |
| Due diligence | Analyse documents, red flags, synthèse | -60% | Faible (analyse préliminaire) |
| Veille réglementaire | Monitoring régulations, impact assessment | -80% | Faible (alerte uniquement) |
| RGPD compliance | Audit DPA, mapping données, registre | -70% | Moyen (audit interne) |
| NDA génération | Templates personnalisés, clauses standard | -90% | Faible (templates validés) |
# /review-contract
Revue automatique de contrat avec scoring de risque.
## Workflow
1. **Lire** le contrat (PDF via Read tool)
2. **Extraire** les clauses clés :
- Durée et renouvellement
- Conditions de résiliation
- Limitation de responsabilité
- Propriété intellectuelle
- Confidentialité
- Juridiction et droit applicable
- SLA et pénalités
- Clauses de non-concurrence
- Force majeure
- Conditions de paiement
3. **Scorer** chaque clause (1-10) :
- 1-3 : Favorable
- 4-6 : Standard / Acceptable
- 7-8 : Risqué / À négocier
- 9-10 : Bloquant / Deal-breaker
4. **Comparer** avec nos standards internes
5. **Générer** le rapport de revue :
- Score global de risque
- Clauses problématiques avec explication
- Suggestions de modification
- Comparaison avec le marché
6. **Notifier** le juriste avec le rapport
## Clauses deal-breaker (auto-flag)
- Responsabilité illimitée
- IP transfer sans compensation
- Non-concurrence > 24 mois
- Juridiction hors EU sans arbitrage
- Pénalités disproportionnées (> 200% du contrat)
- Clause d'audit sans préavis#!/bin/bash
# Veille réglementaire quotidienne
# Monitore les changements RGPD, DORA, AI Act, SOX
claude -p "Effectue la veille réglementaire quotidienne :
1. Recherche web : nouveaux textes et guidelines pour RGPD, DORA, AI Act, NIS2
2. Recherche web : décisions récentes de la CNIL, EDPB, SEC
3. Pour chaque nouveauté :
- Résumé en 3 lignes
- Impact pour notre entreprise (nul, faible, moyen, fort)
- Date d'entrée en vigueur
- Action requise (aucune, veille, mise en conformité, urgente)
4. Génère un rapport synthétique
5. Si impact fort ou action urgente : alerte Slack #legal + email DPO
Format : JSON structuré + rapport HTML" \
--output-format json > /tmp/regulatory-watch-$(date +%Y%m%d).jsonLes Ressources Humaines sont un département paradoxal : ils gèrent le capital le plus précieux de l'entreprise (les humains), mais passent 60% de leur temps sur de l'administratif. Claude Code peut automatiser la partie transactionnelle pour permettre aux RH de se concentrer sur la stratégie talent, la culture d'entreprise et le développement des compétences.
"""
MCP Server : Recrutement assisté par IA
Screening CV, scoring, préparation d'entretien.
"""
import json
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("hr-recruitment")
@mcp.tool()
async def screen_cv(
cv_text: str,
job_requirements: dict
) -> str:
"""Analyse un CV et le score contre les exigences du poste.
Args:
cv_text: Texte extrait du CV
job_requirements: {
"title": str,
"must_have": ["Python", "5+ ans expérience"],
"nice_to_have": ["Kubernetes", "Team lead"],
"education": "Bac+5 ou équivalent",
"languages": ["Français", "Anglais"]
}
"""
return json.dumps({
"instructions": "Analysez ce CV :\n\n"
f"EXIGENCES : {json.dumps(job_requirements)}\n\n"
f"CV : {cv_text[:3000]}\n\n"
"Retournez :\n"
"- match_score (0-100)\n"
"- must_have_met: [{skill, found: bool, evidence}]\n"
"- nice_to_have_met: [{skill, found: bool}]\n"
"- experience_years: int\n"
"- education_match: bool\n"
"- red_flags: [str] (gaps, incohérences)\n"
"- strengths: [str]\n"
"- recommendation: INTERVIEW / MAYBE / PASS\n\n"
"IMPORTANT : Ne PAS discriminer sur âge, genre, origine, handicap."
})
@mcp.tool()
async def generate_job_description(
title: str,
department: str,
level: str,
key_responsibilities: list[str],
tech_stack: list[str] = None,
company_values: list[str] = None
) -> str:
"""Génère une fiche de poste attractive et inclusive."""
return json.dumps({
"instructions": f"Générez une fiche de poste pour : {title}\n"
f"Département : {department}\n"
f"Niveau : {level}\n"
f"Responsabilités : {key_responsibilities}\n"
f"Stack : {tech_stack}\n\n"
"Structure :\n"
"1. Accroche engageante (pas de jargon corporate)\n"
"2. Mission du poste (3 lignes)\n"
"3. Responsabilités principales (5-7 points)\n"
"4. Profil recherché (must-have vs nice-to-have)\n"
"5. Ce que nous offrons\n"
"6. Processus de recrutement\n\n"
"Règles :\n"
"- Langage inclusif (pas de 'ninja', 'rockstar')\n"
"- Fourchette salariale obligatoire\n"
"- Avantages concrets (pas de 'environnement dynamique')\n"
"- Maximum 10 must-have (pas de liste de 25 compétences)"
})
@mcp.tool()
async def prepare_interview(
candidate_cv: str,
job_title: str,
interview_type: str = "technical"
) -> str:
"""Prépare un guide d'entretien personnalisé pour un candidat.
Types : technical, behavioral, cultural, final."""
return json.dumps({
"instructions": f"Préparez un guide d'entretien {interview_type} pour :\n"
f"Poste : {job_title}\n"
f"CV du candidat : {candidate_cv[:2000]}\n\n"
"Générez :\n"
"1. 5 questions techniques ciblées (basées sur l'expérience du candidat)\n"
"2. 3 questions comportementales (STAR method)\n"
"3. 2 mises en situation pratiques\n"
"4. Points spécifiques à vérifier (basés sur les red flags du CV)\n"
"5. Grille d'évaluation (critères + notation 1-5)\n\n"
"IMPORTANT : Questions légales uniquement. Pas de questions sur "
"âge, situation familiale, religion, origines, santé."
})
if __name__ == "__main__":
mcp.run(transport="stdio")#!/bin/bash
# Analyse hebdomadaire du sentiment employé
# Sources : Slack (anonymisé), surveys, 1:1 notes
claude -p "Analyse le sentiment employé cette semaine :
1. Analyse les résultats du pulse survey hebdomadaire (score NPS, commentaires)
2. Détecte les tendances : thèmes récurrents, équipes en difficulté
3. Compare avec les 4 dernières semaines
4. Identifie les signaux faibles :
- Baisse de participation aux événements
- Augmentation des demandes de congés
- Turnover risk par département
5. Recommandations d'actions (prioritées)
6. Génère un rapport pour le CHRO
IMPORTANT : Toutes les données sont ANONYMISÉES.
NE JAMAIS identifier un employé spécifique.
Focus sur les tendances d'équipe, pas les individus." \
--output-format jsonAu-delà de la comptabilité (section 3), les finance operations couvrent la gestion de trésorerie, l'exposition devises, l'analyse budgétaire et la reconnaissance du revenu. Ces processus sont critiques, souvent manuels, et bénéficient énormément de l'automatisation intelligente.
"""
MCP Server : Treasury Management
Prévision de cash flow, gestion FX, analyse de variance budgétaire.
"""
import json
from datetime import datetime, timedelta
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("treasury")
@mcp.tool()
async def cash_flow_forecast(
horizon_days: int = 90,
scenario: str = "base"
) -> str:
"""Génère une prévision de cash flow à N jours.
Scénarios : base, optimiste, pessimiste, stress.
Sources de données :
- Factures clients en attente (AR aging)
- Factures fournisseurs planifiées (AP aging)
- Salaires et charges fixes
- Remboursements de dette
- Revenue forecast (pipeline CRM)
- Cash actuel en banque
"""
return json.dumps({
"instructions": f"Générez un forecast de cash flow à {horizon_days} jours :\n\n"
"1. Récupérez le solde bancaire actuel (via MCP accounting)\n"
"2. Récupérez l'AR aging (factures clients en attente)\n"
"3. Récupérez l'AP aging (factures fournisseurs à payer)\n"
"4. Intégrez les charges fixes (salaires, loyer, SaaS)\n"
"5. Intégrez le revenue forecast du CRM\n"
f"6. Scénario : {scenario}\n"
" - base : paiements aux dates prévues, conversion pipeline 30%\n"
" - optimiste : paiements anticipés, conversion 50%\n"
" - pessimiste : retards de paiement +30j, conversion 15%\n"
" - stress : 2 plus gros clients en retard, pas de new revenue\n\n"
"Output : tableau jour par jour avec cumul, graphique, alertes si cash < seuil"
})
@mcp.tool()
async def fx_exposure_analysis(
base_currency: str = "EUR",
threshold_pct: float = 5.0
) -> str:
"""Analyse l'exposition aux devises étrangères.
Signal les expositions > threshold_pct du cash total."""
return json.dumps({
"instructions": "Analysez l'exposition FX :\n"
"1. Listez tous les comptes par devise\n"
"2. Listez les factures AR/AP par devise\n"
"3. Calculez l'exposition nette par devise\n"
"4. Appliquez les taux de change actuels\n"
"5. Identifiez les expositions significatives\n"
f"6. Seuil d'alerte : {threshold_pct}% du cash total\n"
"7. Recommandations de couverture (hedging)\n\n"
"Devises à surveiller : USD, GBP, CHF, JPY, CAD"
})
@mcp.tool()
async def budget_variance_analysis(
period: str = "this_month",
department: str = None
) -> str:
"""Analyse des écarts budgétaires : réel vs budget vs forecast."""
dept_filter = f"Département : {department}" if department else "Tous les départements"
return json.dumps({
"instructions": f"Analyse de variance budgétaire pour {period} :\n"
f"{dept_filter}\n\n"
"1. Récupérez les dépenses réelles via le Semantic Layer\n"
"2. Récupérez le budget alloué\n"
"3. Calculez les écarts (absolu et %)\n"
"4. Classifiez : favorable, défavorable, neutre\n"
"5. Top 5 postes avec le plus gros écart\n"
"6. Explication probable de chaque écart significatif (>10%)\n"
"7. Forecast ajusté pour le reste de l'année\n"
"8. Recommandations d'optimisation"
})
@mcp.tool()
async def expense_report_process(
employee_id: str,
expenses: list[dict]
) -> str:
"""Traitement automatisé des notes de frais.
Vérification politique, classification, approbation workflow."""
return json.dumps({
"instructions": f"Traitez la note de frais de l'employé {employee_id} :\n\n"
f"Dépenses : {json.dumps(expenses)}\n\n"
"Pour chaque dépense :\n"
"1. Vérifiez la conformité à la politique de dépenses :\n"
" - Repas : max 25€/personne (déjeuner), 50€/personne (dîner)\n"
" - Transport : train/avion préféré au taxi\n"
" - Hôtel : max 180€/nuit (Paris), 150€/nuit (province)\n"
" - Justificatif obligatoire > 25€\n"
"2. Classifiez (compte comptable)\n"
"3. Détectez les anomalies (doublons, montants inhabituels)\n"
"4. Route pour approbation :\n"
" - < 200€ : auto-apprové\n"
" - 200-1000€ : manager\n"
" - > 1000€ : CFO"
})
if __name__ == "__main__":
mcp.run(transport="stdio")Déployer une IA au cœur de chaque processus métier crée des opportunités formidables — et des risques considérables. Cette section est la plus critique de ce guide. Sans une gouvernance solide, l'entreprise autonome est une bombe à retardement.
| Niveau | Exemples | Accès Claude | Contrôles |
|---|---|---|---|
| CONFIDENTIEL | Salaires, contrats, stratégie M&A | Interdit sans approbation explicite | Chiffrement, ACL, audit complet |
| RESTREINT | Données clients, factures, RH | Lecture seule, jamais en log | PII masking, accès temporaire |
| INTERNE | KPIs, rapports, procédures | Lecture et analyse | Authentification, logging |
| PUBLIC | Site web, docs produit, blog | Accès libre | Aucun contrôle spécial |
{
"hooks": {
"PreToolUse": [
{
"matcher": "mcp__accounting__*",
"hooks": [{
"type": "command",
"command": "python /security/check_accounting_auth.py \"$TOOL_INPUT\"",
"description": "Vérifie l'autorisation pour les opérations comptables"
}]
},
{
"matcher": "mcp__hr__*",
"hooks": [{
"type": "command",
"command": "python /security/check_pii_access.py \"$TOOL_INPUT\"",
"description": "Vérifie que les données PII sont nécessaires et autorisées"
}]
},
{
"matcher": "Bash",
"hooks": [{
"type": "command",
"command": "python /security/sandbox_check.py \"$TOOL_INPUT\"",
"description": "Vérifie que la commande est dans la whitelist"
}]
}
],
"PostToolUse": [
{
"matcher": "*",
"hooks": [{
"type": "command",
"command": "python /security/audit_log.py \"$TOOL_NAME\" \"$TOOL_INPUT\" \"$TOOL_OUTPUT\"",
"description": "Log chaque action dans l'audit trail"
}]
}
]
}
}"""
Audit trail pour toutes les actions de Claude Code.
Chaque appel d'outil est loggé avec : timestamp, outil, input, output,
utilisateur, session, durée, statut.
"""
import json
import sys
import hashlib
from datetime import datetime, timezone
def log_action(tool_name: str, tool_input: str, tool_output: str):
"""Enregistre une action dans l'audit trail."""
# Masquer les données sensibles
sanitized_input = mask_pii(tool_input)
sanitized_output = mask_pii(tool_output)
entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"tool": tool_name,
"input_hash": hashlib.sha256(tool_input.encode()).hexdigest()[:16],
"input_preview": sanitized_input[:500],
"output_preview": sanitized_output[:500],
"session_id": get_session_id(),
"user": get_current_user(),
"status": "success" if "error" not in tool_output.lower() else "error",
"data_classification": classify_data_level(tool_name, tool_input)
}
# Écrire dans le fichier d'audit (append-only)
with open("/var/log/claude-audit/audit.jsonl", "a") as f:
f.write(json.dumps(entry) + "\n")
# Si données CONFIDENTIELLES : alerte immédiate
if entry["data_classification"] == "CONFIDENTIEL":
send_security_alert(entry)
def mask_pii(text: str) -> str:
"""Masque les données personnelles dans les logs."""
import re
# Emails
text = re.sub(r'[\w.-]+@[\w.-]+\.\w+', '[EMAIL_MASKED]', text)
# Téléphones
text = re.sub(r'\b\d{2}[\s.-]?\d{2}[\s.-]?\d{2}[\s.-]?\d{2}[\s.-]?\d{2}\b',
'[PHONE_MASKED]', text)
# IBAN
text = re.sub(r'\b[A-Z]{2}\d{2}[\s]?[\dA-Z]{4}[\s]?[\dA-Z]{4}[\s]?[\dA-Z]{4}',
'[IBAN_MASKED]', text)
# SSN / SIRET
text = re.sub(r'\b\d{3}[\s.-]?\d{3}[\s.-]?\d{3}[\s.-]?\d{3,5}\b',
'[ID_MASKED]', text)
return text
def classify_data_level(tool_name: str, tool_input: str) -> str:
"""Classifie le niveau de sensibilité de l'opération."""
confidential_tools = ["hr_salary", "hr_personal", "legal_contract", "treasury_transfer"]
restricted_tools = ["accounting", "support_customer", "hr_recruitment"]
for t in confidential_tools:
if t in tool_name.lower():
return "CONFIDENTIEL"
for t in restricted_tools:
if t in tool_name.lower():
return "RESTREINT"
return "INTERNE"
def get_session_id() -> str:
import os
return os.environ.get("CLAUDE_SESSION_ID", "unknown")
def get_current_user() -> str:
import os
return os.environ.get("USER", "system")
def send_security_alert(entry: dict):
"""Envoie une alerte sécurité pour les accès CONFIDENTIELS."""
import subprocess
subprocess.run(["curl", "-X", "POST", "-H", "Content-type: application/json",
"--data", json.dumps({
"text": f":rotating_light: ACCÈS CONFIDENTIEL\n"
f"Outil: {entry['tool']}\n"
f"User: {entry['user']}\n"
f"Session: {entry['session_id']}\n"
f"Timestamp: {entry['timestamp']}"
}),
"https://hooks.slack.com/services/T.../B.../security"],
capture_output=True)
if __name__ == "__main__":
log_action(sys.argv[1], sys.argv[2], sys.argv[3])| Action | Seuil d'approbation | Approbateur | SLA |
|---|---|---|---|
| Création de facture | Toujours (DRAFT) | Comptable | 24h |
| Envoi d'email prospect | Toujours (review) | Commercial | 4h |
| Virement > 5 000€ | Toujours | CFO + CEO | 2h |
| Modification contrat | Toujours | Juriste | 48h |
| Réponse support L2+ | Client Enterprise | CSM Senior | 1h |
| Déploiement prod | Toujours | Tech Lead + QA | 30 min |
| Accès données RH | PII direct | DPO | 4h |
| Rapport board | Toujours | CFO | 24h |
| Catégorie | Vérification | Statut |
|---|---|---|
| Authentification | Chaque MCP server a ses propres credentials (pas de token partagé) | ☐ |
| Autorisation | Principe du moindre privilège sur chaque MCP server | ☐ |
| Chiffrement | TLS pour tous les transports MCP (SSE, HTTP) | ☐ |
| Audit | Logging de chaque action via PostToolUse hook | ☐ |
| PII | Masking des données personnelles dans les logs | ☐ |
| Rate limiting | Limites par MCP server (ex: max 100 requêtes/heure comptabilité) | ☐ |
| Cost control | Budget max par session Claude ($50/jour, $500/mois) | ☐ |
| Sandbox | Commandes Bash dans une whitelist (pas de rm, pas de sudo) | ☐ |
| Backup | Toutes les décisions sont réversibles (soft delete, versioning) | ☐ |
| Testing | Chaque workflow testé en staging avant prod | ☐ |
| Monitoring | Alertes Grafana sur erreurs, latence, coûts | ☐ |
| Incident | Procédure de kill switch (arrêt d'urgence de tous les MCP servers) | ☐ |
"""
Contrôle des coûts Claude Code en entreprise.
Limites par utilisateur, par département, par workflow.
"""
import json
from datetime import datetime, timezone
LIMITS = {
"daily_per_user": 50.0, # $50/jour/utilisateur
"monthly_per_dept": 5000.0, # $5000/mois/département
"per_session": 20.0, # $20/session max
"monthly_total": 25000.0 # $25K/mois total entreprise
}
def check_budget(user: str, department: str, estimated_cost: float) -> dict:
"""Vérifie si l'opération est dans le budget."""
# Récupérer les coûts actuels (depuis la DB de tracking)
current_daily = get_daily_spend(user)
current_monthly_dept = get_monthly_dept_spend(department)
current_monthly_total = get_monthly_total_spend()
checks = {
"daily_user": {
"current": current_daily,
"limit": LIMITS["daily_per_user"],
"remaining": LIMITS["daily_per_user"] - current_daily,
"ok": current_daily + estimated_cost <= LIMITS["daily_per_user"]
},
"monthly_dept": {
"current": current_monthly_dept,
"limit": LIMITS["monthly_per_dept"],
"remaining": LIMITS["monthly_per_dept"] - current_monthly_dept,
"ok": current_monthly_dept + estimated_cost <= LIMITS["monthly_per_dept"]
},
"monthly_total": {
"current": current_monthly_total,
"limit": LIMITS["monthly_total"],
"remaining": LIMITS["monthly_total"] - current_monthly_total,
"ok": current_monthly_total + estimated_cost <= LIMITS["monthly_total"]
}
}
all_ok = all(c["ok"] for c in checks.values())
return {
"approved": all_ok,
"checks": checks,
"action": "PROCEED" if all_ok else "BLOCK",
"message": "Budget OK" if all_ok else
"Budget dépassé - contactez votre manager pour extension"
}Construire une entreprise autonome nécessite un stack technologique cohérent où chaque brique communique avec les autres. Voici l'architecture de référence, testée en production, avec les alternatives par composant.
┌─────────────────────────────────────────────────────────────────────────┐
│ COUCHE PRESENTATION │
│ Slack Bot · Email · Notion · Dashboard Grafana · Reports HTML/PDF │
└───────────────────────────┬─────────────────────────────────────────────┘
│
┌───────────────────────────┼─────────────────────────────────────────────┐
│ COUCHE INTELLIGENCE │
│ │
│ ┌──────────────────────────────────────────────────────────────────┐ │
│ │ CLAUDE CODE CENTRAL │ │
│ │ Skills · Raisonnement · Décision · Génération · Orchestration │ │
│ └──────────────────────────────────────────────────────────────────┘ │
│ │ │ │ │ │ │ │
│ ┌────┴──┐ ┌────┴──┐ ┌───┴───┐ ┌───┴───┐ ┌──┴───┐ ┌──┴───┐ │
│ │ Hooks │ │ CLAUDE│ │ MCP │ │ MCP │ │ MCP │ │ MCP │ │
│ │ Sécu │ │ .md │ │Compta │ │ CRM │ │ RH │ │ Data │ │
│ │ Audit │ │ Skills│ │ Xero │ │HubSpot│ │ SIRH │ │ SL │ │
│ └───────┘ └───────┘ └───────┘ └───────┘ └──────┘ └──────┘ │
└───────────────────────────┬─────────────────────────────────────────────┘
│
┌───────────────────────────┼─────────────────────────────────────────────┐
│ COUCHE ORCHESTRATION │
│ │
│ Temporal / n8n Cron / Discord Bot Webhooks / Events │
│ Workflows durables Tâches planifiées Déclencheurs temps réel│
└───────────────────────────┬─────────────────────────────────────────────┘
│
┌───────────────────────────┼─────────────────────────────────────────────┐
│ COUCHE DONNÉES │
│ │
│ ┌────────────┐ ┌───────────────┐ ┌─────────────┐ ┌──────────────┐ │
│ │ PostgreSQL │ │ dbt Core │ │ Semantic │ │ S3 / GCS │ │
│ │ Snowflake │ │ Transform │ │ Layer │ │ Documents │ │
│ │ BigQuery │ │ Tests │ │ MetricFlow │ │ Factures │ │
│ └────────────┘ └───────────────┘ └─────────────┘ └──────────────┘ │
│ │
│ ┌────────────┐ ┌───────────────┐ ┌─────────────┐ ┌──────────────┐ │
│ │ Fivetran │ │ Airbyte │ │ Redis │ │ Vault │ │
│ │ ELT Ingestion│ │ Open-source │ │ Cache │ │ Secrets │ │
│ └────────────┘ └───────────────┘ └─────────────┘ └──────────────┘ │
└───────────────────────────┬─────────────────────────────────────────────┘
│
┌───────────────────────────┼─────────────────────────────────────────────┐
│ COUCHE MONITORING │
│ │
│ Grafana · Prometheus · Audit Logs · Cost Tracking · Alerting │
└─────────────────────────────────────────────────────────────────────────┘
| Composant | Recommandé | Alternative | Pourquoi |
|---|---|---|---|
| Base de données | PostgreSQL | Snowflake, BigQuery | Open-source, extensible, excellent pour dbt |
| Data Warehouse | Snowflake | BigQuery, Redshift | Scaling auto, séparation compute/storage |
| Transform | dbt Core | dbt Cloud, SQLMesh | Standard de facto, gratuit, git-native |
| Semantic Layer | dbt Semantic Layer | Cube.js, Lightdash | Intégration native dbt, MetricFlow |
| IA | Claude Code (Opus 4) | Claude Sonnet 4 | Raisonnement supérieur, MCP natif |
| Orchestration | Temporal | n8n, Inngest | Workflows durables, retry intelligent |
| Ingestion | Fivetran | Airbyte, Stitch | 300+ connecteurs, maintenance zéro |
| Cache | Redis | Memcached | Structures de données riches, Pub/Sub |
| Secrets | HashiCorp Vault | AWS Secrets Manager | Multi-cloud, rotation automatique |
| Monitoring | Grafana + Prometheus | Datadog | Open-source, dashboards customisables |
| Communication | Slack | Teams, Discord | API riche, webhooks natifs, bots |
| CRM | HubSpot | Salesforce, Pipedrive | API moderne, free tier généreux |
| Comptabilité | Xero | QuickBooks, Pennylane | API REST moderne, multi-devises |
| Composant | PME (50 emp.) | Mid-Market (200 emp.) | Enterprise (1000+ emp.) |
|---|---|---|---|
| Claude Code API | $800/mois | $3 500/mois | $15 000/mois |
| Data Warehouse | $200/mois | $1 200/mois | $8 000/mois |
| Ingestion (Fivetran) | $500/mois | $2 000/mois | $10 000/mois |
| Orchestration | $0 (self-hosted) | $500/mois | $3 000/mois |
| SaaS (CRM, Compta) | $300/mois | $1 500/mois | $8 000/mois |
| Monitoring | $0 (open-source) | $300/mois | $2 000/mois |
| Infrastructure | $200/mois | $1 000/mois | $5 000/mois |
| TOTAL | $2 000/mois | $10 000/mois | $51 000/mois |
| Économie estimée | $8 000/mois | $45 000/mois | $200 000/mois |
| ROI | 4x | 4.5x | 3.9x |
L'automatisation IA ne se justifie que si elle génère un retour sur investissement mesurable. Cette section détaille comment calculer, mesurer et démontrer le ROI de chaque automatisation, département par département.
ROI = (Gains - Coûts) / Coûts × 100
Gains =
+ Heures économisées × Coût horaire chargé
+ Erreurs évitées × Coût moyen d'une erreur
+ Revenue additionnel (prospection, conversion)
+ Amélioration qualité (CSAT, NPS)
Coûts =
+ Licences / API Claude Code
+ Infrastructure (hosting, data warehouse)
+ Développement (MCP servers, intégrations)
+ Maintenance (monitoring, updates)
+ Formation (équipes, change management)| Département | Processus automatisé | Temps économisé | Économie annuelle | Time-to-Value |
|---|---|---|---|---|
| Comptabilité | Factures, rapprochement, clôture | 85% du temps répétitif | 120 000 € | 2 mois |
| Commercial | Prospection, enrichissement, CRM | 60% du temps non-vente | 180 000 € | 3 mois |
| Support | Triage, L1 auto, knowledge base | 70% des tickets L1 | 250 000 € | 1 mois |
| Finance | Reporting, trésorerie, budget | 75% du reporting | 95 000 € | 2 mois |
| RH | Screening, onboarding, pulse | 50% de l'administratif | 80 000 € | 3 mois |
| Juridique | Revue contrats, veille, NDA | 60% de la revue initiale | 110 000 € | 4 mois |
| Data | Pipelines, qualité, catalogue | 40% de la maintenance | 70 000 € | 2 mois |
| Ops / IT | Incidents, déploiements, monitoring | 50% des incidents P3/P4 | 90 000 € | 2 mois |
| TOTAL (PME 200 emp.) | 995 000 €/an | 2-4 mois | ||
Transformer une entreprise traditionnelle en entreprise autonome est un voyage de 12 à 18 mois. Voici la roadmap phase par phase, avec les quick wins, les investissements clés et les critères de succès à chaque étape.
Phase 1 — Fondations (Mois 1-2)
================================
[ ] PostgreSQL ou Snowflake déployé et configuré
[ ] dbt Core installé, premier modèle de staging créé
[ ] 5-10 métriques clés définies dans le Semantic Layer
[ ] Claude Code installé, CLAUDE.md écrit
[ ] Premier MCP server (le plus simple : email ou Slack)
[ ] Hooks d'audit déployés (PostToolUse logging)
[ ] Politique de sécurité IA validée par le RSSI
[ ] Classification des données documentée
[ ] Budget IA approuvé (coûts API + infra)
[ ] Équipe formée (2-3 personnes minimum)| Rôle | Phase | Temps | Profil |
|---|---|---|---|
| AI Engineer | Phase 1+ | Full-time | Python, MCP, Claude Code, dbt |
| Data Engineer | Phase 1+ | Full-time | SQL, dbt, Snowflake/Postgres, pipelines |
| Product Owner IA | Phase 2+ | Full-time | Comprend le métier ET la tech, priorise |
| Security Engineer | Phase 1+ | 50% (puis full-time Phase 3) | InfoSec, audit, compliance |
| Champions métier | Phase 2+ | 20% chacun | 1 par département, relais terrain |
| Poste | Année 1 | Année 2 | Notes |
|---|---|---|---|
| Salaires équipe IA | 280 000 € | 350 000 € | 2 ingénieurs + 1 PO (année 1), +1 sécu (année 2) |
| Licences & API | 60 000 € | 90 000 € | Claude Code, Snowflake, Fivetran, SaaS |
| Infrastructure | 24 000 € | 36 000 € | Cloud, monitoring, backup |
| Formation | 15 000 € | 10 000 € | Change management, workshops |
| TOTAL | 379 000 € | 486 000 € | |
| Économies générées | 500 000 € | 995 000 € | Progressif : quick wins dès M3 |
| ROI cumulé | +32% | +105% | Payback : 8-10 mois |