103 lines
3.8 KiB
Python
103 lines
3.8 KiB
Python
from mcp.server.fastmcp import FastMCP
|
|
import chromadb
|
|
import os
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
# 1. Initialisation de FastMCP
|
|
# On force l'hôte sur 0.0.0.0 pour Docker et le port sur 8080
|
|
mcp = FastMCP("Chroma-Admin-Assistant", host="0.0.0.0", port=8080)
|
|
|
|
# 2. Connexion au client ChromaDB
|
|
# On utilise l'URL interne au réseau Docker définie dans votre docker-compose
|
|
CHROMA_HOST = os.getenv("CHROMA_HOST", "chroma-db")
|
|
client = chromadb.HttpClient(host=CHROMA_HOST, port=8000)
|
|
|
|
# Nom de la collection dédiée aux incidents
|
|
COLLECTION_NAME = "incident_history"
|
|
|
|
@mcp.tool()
|
|
async def find_incident_solution(query_error: str, service: str = None):
|
|
"""
|
|
Recherche des incidents passés similaires et retourne les solutions appliquées.
|
|
L'utilisateur peut décrire l'erreur ou donner un extrait de log technique.
|
|
"""
|
|
try:
|
|
collection = client.get_or_create_collection(name=COLLECTION_NAME)
|
|
|
|
# Recherche vectorielle des 2 incidents les plus proches
|
|
results = collection.query(
|
|
query_texts=[query_error],
|
|
n_results=2,
|
|
where={"service": service} if service else None
|
|
)
|
|
|
|
if not results["documents"] or not results["documents"][0]:
|
|
return "Aucun incident similaire répertorié pour le moment."
|
|
|
|
response = "🔍 Voici les précédents incidents similaires trouvés :\n"
|
|
for i in range(len(results["documents"][0])):
|
|
doc = results["documents"][0][i]
|
|
meta = results["metadatas"][0][i]
|
|
response += f"\n--- INCIDENT {meta.get('incident_id')} ---"
|
|
response += f"\nLOGS : {doc}"
|
|
response += f"\nSOLUTION : {meta.get('solution_applied')}\n"
|
|
|
|
return response
|
|
except Exception as e:
|
|
return f"Erreur lors de la recherche : {str(e)}"
|
|
|
|
@mcp.tool()
|
|
async def add_incident_report(service: str, incident_description: str, solution_applied: str):
|
|
"""
|
|
Ajoute un nouvel incident et sa résolution dans la base de connaissances.
|
|
À utiliser quand une solution technique a été validée par un administrateur.
|
|
"""
|
|
try:
|
|
collection = client.get_or_create_collection(name=COLLECTION_NAME)
|
|
|
|
uid = f"INC-{uuid.uuid4().hex[:8].upper()}"
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
collection.add(
|
|
documents=[incident_description],
|
|
metadatas=[{
|
|
"service": service,
|
|
"solution_applied": solution_applied,
|
|
"incident_id": uid,
|
|
"timestamp": timestamp
|
|
}],
|
|
ids=[uid]
|
|
)
|
|
|
|
return f"✅ Succès : L'incident {uid} a été archivé pour le service {service}."
|
|
except Exception as e:
|
|
return f"Erreur lors de l'ajout : {str(e)}"
|
|
|
|
@mcp.tool()
|
|
async def list_all_incidents(limit: int = 15):
|
|
"""
|
|
Liste les incidents enregistrés dans la base.
|
|
Permet d'avoir une vue d'ensemble des connaissances actuelles.
|
|
"""
|
|
try:
|
|
collection = client.get_or_create_collection(name=COLLECTION_NAME)
|
|
results = collection.get(limit=limit)
|
|
|
|
if not results["ids"]:
|
|
return "La base de connaissances est vide."
|
|
|
|
summary = f"📋 Liste des {len(results['ids'])} dernières entrées :\n"
|
|
for i in range(len(results["ids"])):
|
|
meta = results["metadatas"][i]
|
|
summary += f"\n- [{meta.get('timestamp')}] {meta.get('incident_id')} | Service: {meta.get('service')}"
|
|
summary += f"\n Extrait: {results['documents'][i][:75]}..."
|
|
|
|
return summary
|
|
except Exception as e:
|
|
return f"Erreur lors de la lecture : {str(e)}"
|
|
|
|
if __name__ == "__main__":
|
|
# Lancement du serveur en mode SSE
|
|
mcp.run(transport="sse")
|