Intermediate · 2026-06-26
# Construire un agent de réapprovisionnement autonome gouverné pour SAP S/4HANA avec Claude et MCP
## Introduction
La gestion du réapprovisionnement dans les environnements ERP d'entreprise nécessite à la fois une automatisation intelligente et une gouvernance stricte. Ce guide démontre comment construire un agent autonome pour SAP S/4HANA qui utilise Claude via le Model Context Protocol (MCP) pour prendre des décisions de réapprovisionnement tout en maintenant des garde-fous d'entreprise.
## Architecture de l'agent
### Composants principaux
L'agent se compose de quatre couches :
1. **Couche MCP** : Expose les fonctions SAP et les données de réapprovisionnement à Claude
2. **Moteur de décision** : Claude analyse les niveaux de stock, la demande et les contraintes
3. **Cadre de gouvernance** : Valide les décisions selon les règles métier
4. **Couche d'exécution** : Exécute les commandes d'achat approuvées dans S/4HANA
### Flux de données
```
SAP S/4HANA → Serveur MCP → Claude → Validateur de gouvernance → SAP S/4HANA
```
## Mise en œuvre du serveur MCP
### Définir les outils de réapprovisionnement
Créez des outils MCP qui exposent les fonctions de réapprovisionnement critiques :
```python
from mcp.server import Server
from mcp.types import Tool, TextContent
import sapnwrfc
app = Server("sap-replenishment-mcp")
@app.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="get_inventory_status",
description="Récupère les niveaux de stock actuels, le stock de sécurité et le délai pour un SKU",
inputSchema={
"type": "object",
"properties": {
"material_number": {"type": "string"},
"plant": {"type": "string"}
},
"required": ["material_number", "plant"]
}
),
Tool(
name="get_demand_forecast",
description="Récupère les prévisions de demande pour un SKU sur un horizon temporel",
inputSchema={
"type": "object",
"properties": {
"material_number": {"type": "string"},
"plant": {"type": "string"},
"horizon_days": {"type": "integer"}
},
"required": ["material_number", "plant"]
}
),
Tool(
name="get_supplier_info",
description="Récupère les informations fournisseur incluant MOQ, prix et performance de livraison",
inputSchema={
"type": "object",
"properties": {
"material_number": {"type": "string"},
"vendor": {"type": "string"}
},
"required": ["material_number"]
}
),
Tool(
name="create_purchase_requisition",
description="Crée une demande d'achat pour approbation (n'exécute pas)",
inputSchema={
"type": "object",
"properties": {
"material_number": {"type": "string"},
"quantity": {"type": "number"},
"plant": {"type": "string"},
"vendor": {"type": "string"},
"delivery_date": {"type": "string"},
"justification": {"type": "string"}
},
"required": ["material_number", "quantity", "plant", "justification"]
}
)
]
```
### Implémenter les appels SAP
Connectez les outils MCP aux fonctions SAP via RFC :
```python
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
conn = sapnwrfc.Connection(
user=SAP_USER,
passwd=SAP_PASS,
ashost=SAP_HOST,
sysnr=SAP_SYSNR,
client=SAP_CLIENT
)
if name == "get_inventory_status":
result = conn.call('BAPI_MATERIAL_STOCK_REQ_LIST', {
'MATERIAL': arguments['material_number'],
'PLANT': arguments['plant']
})
inventory_data = {
'current_stock': result['STOCK_QUANTITY'],
'safety_stock': result['SAFETY_STOCK'],
'reorder_point': result['REORDER_POINT'],
'lead_time_days': result['LEAD_TIME'],
'in_transit': result['IN_TRANSIT_QTY']
}
return [TextContent(
type="text",
text=json.dumps(inventory_data, indent=2)
)]
elif name == "get_demand_forecast":
result = conn.call('BAPI_DEMAND_FORECAST_GET', {
'MATERIAL': arguments['material_number'],
'PLANT': arguments['plant'],
'HORIZON': arguments.get('horizon_days', 90)
})
return [TextContent(
type="text",
text=json.dumps(result['FORECAST_DATA'], indent=2)
)]
# Implémenter d'autres outils de manière similaire...
```
## Invites d'agent pour Claude
### Invite système
Définissez la personnalité et les contraintes de l'agent :
```python
SYSTEM_PROMPT = """Vous êtes un agent de réapprovisionnement autonome pour SAP S/4HANA. Votre objectif est de maintenir des niveaux de stock optimaux tout en minimisant les coûts et en prévenant les ruptures de stock.
Responsabilités :
- Surveiller les niveaux de stock à travers les usines
- Analyser les prévisions de demande et les tendances
- Recommander des quantités de réapprovisionnement en tenant compte de MOQ, délais et contraintes budgétaires
- Fournir une justification claire pour chaque décision
Contraintes obligatoires :
1. Ne jamais dépasser les limites d'approbation d'achat (50 000€ par réquisition)
2. Respecter les contrats fournisseurs et MOQ
3. Maintenir le stock de sécurité au-dessus des niveaux minimum
4. Tenir compte de la saisonnalité et des tendances de demande
5. Vérifier le stock en transit avant de commander
6. Expliquer toutes les décisions avec des données
Vous avez accès aux outils SAP via MCP pour interroger l'inventaire, les prévisions et les informations fournisseurs. Utilisez toujours des données pour soutenir vos recommandations.
"""
```
### Invite de tâche
Créez des invites structurées pour les tâches de réapprovisionnement :
```python
def create_replenishment_task(material_number: str, plant: str) -> str:
return f"""Analyser les besoins de réapprovisionnement pour :
- Numéro de matériel : {material_number}
- Usine : {plant}
Processus :
1. Récupérer le statut actuel de l'inventaire
2. Obtenir les prévisions de demande pour les 90 prochains jours
3. Vérifier le stock en transit
4. Examiner les informations et performance des fournisseurs
5. Calculer la quantité de commande optimale
6. Vérifier si le réapprovisionnement est nécessaire
Si le réapprovisionnement est nécessaire :
- Déterminer la quantité optimale (considérer MOQ, remises par paliers)
- Sélectionner le meilleur fournisseur
- Calculer la date de livraison requise
- Créer une demande d'achat avec justification détaillée
Fournir le raisonnement complet pour votre décision.
"""
```
## Cadre de gouvernance
### Validateur de règles
Implémentez des vérifications de validation avant l'exécution :
```python
class GovernanceValidator:
def __init__(self, config: dict):
self.max_purchase_value = config['max_purchase_value']
self.approved_vendors = config['approved_vendors']
self.required_approvers = config['required_approvers']
def validate_requisition(self, requisition: dict) -> tuple[bool, list[str]]:
"""Valide la demande d'achat selon les règles de gouvernance"""
errors = []
# Vérifier les limites de valeur
total_value = requisition['quantity'] * requisition['unit_price']
if total_value > self.max_purchase_value:
errors.append(
f"Valeur de réquisition {total_value}€ dépasse la limite {self.max_purchase_value}€. "
"Approbation de niveau supérieur requise."
)
# Vérifier les fournisseurs approuvés
if requisition['vendor'] not in self.approved_vendors:
errors.append(
f"Fournisseur {requisition['vendor']} non dans la liste approuvée. "
"Approbation d'achat requise."
)
# Vérifier le stock de sécurité
if requisition['resulting_stock'] < requisition['safety_stock']:
errors.append(
"La quantité commandée ne maintiendra pas le stock de sécurité. "
"Augmenter la quantité de commande."
)
# Vérifier le MOQ
if requisition['quantity'] < requisition['moq']:
errors.append(
f"Quantité {requisition['quantity']} inférieure au MOQ {requisition['moq']}. "
"Ajuster à la quantité minimum."
)
# Vérifier les contraintes budgétaires
budget_available = self.get_budget_available(
requisition['plant'],
requisition['cost_center']
)
if total_value > budget_available:
errors.append(
f"Fonds insuffisants. Requis : {total_value}€, Disponible : {budget_available}€"
)
return len(errors) == 0, errors
def get_budget_available(self, plant: str, cost_center: str) -> float:
"""Interroge SAP pour le budget disponible"""
# Implémenter l'appel SAP pour vérifier le budget
pass
```
### Flux d'approbation
Implémentez des approbations multi-niveaux basées sur la valeur :
```python
class ApprovalWorkflow:
def __init__(self):
self.approval_levels = {
'L1': {'max_value': 10000, 'approvers': ['procurement_specialist']},
'L2': {'max_value': 50000, 'approvers': ['procurement_manager']},
'L3': {'max_value': float('inf'), 'approvers': ['cfo', 'procurement_director']}
}
def get_required_approvers(self, requisition_value: float) -> list[str]:
"""Détermine les approbateurs requis en fonction de la valeur"""
for level, config in self.approval_levels.items():
if requisition_value <= config['max_value']:
return config['approvers']
return self.approval_levels['L3']['approvers']
def submit_for_approval(self, requisition: dict, approvers: list[str]):
"""Soumet la réquisition pour approbation humaine"""
approval_request = {
'requisition_id': requisition['id'],
'material': requisition['material_number'],
'quantity': requisition['quantity'],
'value': requisition['total_value'],
'vendor': requisition['vendor'],
'justification': requisition['justification'],
'agent_reasoning': requisition['agent_reasoning'],
'approvers': approvers,
'created_at': datetime.now().isoformat()
}
# Envoyer aux approbateurs via système de workflow
self.send_approval_notification(approval_request)
return approval_request['requisition_id']
```
## Orchestrateur de l'agent
### Boucle d'exécution principale
Orchestrez le cycle complet de décision de l'agent :
```python
from anthropic import Anthropic
class ReplenishmentAgent:
def __init__(self, mcp_server_url: str, governance_config: dict):
self.client = Anthropic()
self.mcp_url = mcp_server_url
self.validator = GovernanceValidator(governance_config)
self.workflow = ApprovalWorkflow()
async def analyze_and_replenish(self, material_number: str, plant: str):
"""Exécute l'analyse de réapprovisionnement et la génération de décision"""
# Créer la tâche pour Claude
task_prompt = create_replenishment_task(material_number, plant)
# Exécuter la conversation Claude avec accès MCP
response = self.client.messages.create(
model="claude-3-7-sonnet-20250219",
max_tokens=4000,
system=SYSTEM_PROMPT,
messages=[{
"role": "user",
"content": task_prompt
}],
tools=await self.get_mcp_tools()
)
# Traiter la décision de l'agent
decision = self.extract_decision(response)
if decision['action'] == 'replenish':
# Valider selon les règles de gouvernance
is_valid, errors = self.validator.validate_requisition(
decision['requisition']
)
if not is_valid:
self.log_governance_violation(decision, errors)
return {
'status': 'rejected',
'reason': errors,
'decision': decision
}
# Déterminer le flux d'approbation
approvers = self.workflow.get_required_approvers(
decision['requisition']['total_value']
)
# Soumettre pour approbation
approval_id = self.workflow.submit_for_approval(
decision['requisition'],
approvers
)
return {
'status': 'pending_approval',
'approval_id': approval_id,
'approvers': approvers,
'decision': decision
}
else:
return {
'status': 'no_action_needed',
'reasoning': decision['reasoning']
}
def extract_decision(self, response) -> dict:
"""Extrait la décision structurée de la réponse de Claude"""
# Parcourir les appels d'outils et analyser la réponse finale
# Extraire les détails de la réquisition si créée
pass
async def get_mcp_tools(self) -> list:
"""Récupère les définitions d'outils du serveur MCP"""
# Se connecter au serveur MCP et lister les outils disponibles
pass
```
## Capacités de surveillance et d'audit
### Journal des décisions
Enregistrez toutes les décisions de l'agent pour audit :
```python
class DecisionLogger:
def __init__(self, db_connection):
self.db = db_connection
def log_decision(self, decision: dict, context: dict):
"""Enregistre les décisions de l'agent avec le contexte complet"""
log_entry = {
'timestamp': datetime.now().isoformat(),
'material_number': context['material_number'],
'plant': context['plant'],
'decision_type': decision['action'],
'reasoning': decision['reasoning'],
'data_used': decision['data_sources'],
'requisition_details': decision.get('requisition'),
'governance_checks': decision['governance_validation'],
'agent_model': 'claude-3-7-sonnet-20250219',
'approval_status': decision.get('approval_status')
}
self.db.insert('agent_decisions', log_entry)
def get_decision_history(self, material_number: str, days: int = 90):
"""Récupère l'historique des décisions pour analyse"""
return self.db.query(
"""
SELECT * FROM agent_decisions
WHERE material_number = ?
AND timestamp > datetime('now', '-' || ? || ' days')
ORDER BY timestamp DESC
""",
(material_number, days)
)
```
### Tableaux de bord de performance
Suivez les KPI de l'agent :
```python
class AgentMetrics:
def __init__(self):
self.metrics = {
'decisions_made': 0,
'requisitions_created': 0,
'governance_violations': 0,
'approval_rate': 0.0,
'avg_decision_time': 0.0,
'stockout_prevention': 0,
'cost_savings': 0.0
}
def calculate_performance(self, time_period: str) -> dict:
"""Calcule les métriques de performance de l'agent"""
decisions = self.get_decisions(time_period)
return {
'total_decisions': len(decisions),
'auto_approved': len([d for d in decisions if d['auto_approved']]),
'human_reviewed': len([d for d in decisions if not d['auto_approved']]),
'rejection_rate': self.calculate_rejection_rate(decisions),
'avg_order_accuracy': self.calculate_order_accuracy(decisions),
'inventory_turns_improvement': self.calculate_turns_improvement(decisions),
'cost_per_decision': self.calculate_cost_efficiency(decisions)
}
```
## Déploiement en production
### Configuration de l'environnement
```yaml
# config.yaml
sap_connection:
host: "sap-prod.company.com"
client: "100"
system_number: "00"
mcp_server:
host: "localhost"
port: 8080
agent:
model: "claude-3-7-sonnet-20250219"
max_tokens: 4000
temperature: 0.0 # Déterministe pour les décisions de production
governance:
max_purchase_value: 50000
approved_vendors_file: "vendors.json"
budget_check_enabled: true
require_human_approval_over: 10000
monitoring:
log_level: "INFO"
metrics_export: true
audit_retention_days: 2555
```
### Considérations de sécurité
```python
class SecurityLayer:
def __init__(self):
self.encryption = AES256Encryption()
self.audit_log = AuditLogger()
def secure_sap_credentials(self, credentials: dict) -> dict:
"""Chiffre les identifiants SAP au repos"""
return {
'user': self.encryption.encrypt(credentials['user']),
'password': self.encryption.encrypt(credentials['password']),
'encrypted': True
}
def validate_tool_access(self, tool_name: str, user_context: dict) -> bool:
"""Valide que l'agent peut accéder à l'outil SAP"""
self.audit_log.log_access_attempt(tool_name, user_context)
# Vérifier les permissions basées sur les rôles
if tool_name == "create_purchase_requisition":
return user_context['role'] in ['procurement_agent', 'admin']
return True
```
## Gestion des cas limites
### Gestion des données manquantes
```python
def handle_missing_data(material_number: str, plant: str) -> dict:
"""Gestion gracieuse lorsque les données SAP sont incomplètes"""
try:
inventory = get_inventory_status(material_number, plant)
except DataNotFoundError:
# Revenir aux paramètres par défaut ou escalader
return {
'action': 'escalate',
'reason': 'Données d'inventaire manquantes pour décision',
'requires_human': True
}
if inventory['forecast_confidence'] < 0.6:
# Faible confiance - utiliser une approche conservatrice
return {
'action': 'conservative_replenish',
'quantity': inventory['safety_stock'] * 1.5,
'reason': 'Faible confiance de prévision - stock de sécurité augmenté'
}
return {'action': 'proceed'}
```
### Scénarios de récupération
```python
class RecoveryHandler:
def handle_sap_connection_failure(self, error: Exception):
"""Gérer les pannes de connexion SAP"""
self.logger.error(f"Échec de connexion SAP : {error}")
# Essayer le système de secours
if self.failover_available():
return self.connect_to_failover()
# Passer en mode lecture seule
self.switch_to_readonly_mode()
self.notify_operations_team()
def handle_governance_system_failure(self):
"""Lorsque le système de gouvernance est en panne"""
# Arrêter toutes les créations de réquisitions automatiques
self.pause_autonomous_actions()
# Passer à l'approbation manuelle uniquement
self.require_human_approval_all()
# Alerter l'équipe de gouvernance
self.send_alert("Système de gouvernance en panne - approbation manuelle requise")
```
## Conclusion
Cet agent de réapprovisionnement autonome démontre comment combiner les capacités de raisonnement de Claude avec les garde-fous de gouvernance d'entreprise via MCP. Les éléments clés incluent :
1. **Exposition structurée des données** via les outils MCP pour SAP
2. **Prise de décision guidée** par des invites système claires
3. **Validation multi-couches** via les cadres de gouvernance
4. **Supervision humaine** pour les décisions à enjeux élevés
5. **Traçabilité complète** via les journaux de décision et d'audit
Cette architecture peut s'étendre à d'autres processus ERP autonomes tout en maintenant le contrôle et la conformité d'entreprise.
