feat(mcp): expose les 4 fonctions métier via @mcp_enabled
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,86 @@
|
|||||||
|
from dash.mcp import mcp_enabled
|
||||||
|
|
||||||
|
from src.mcp import queries
|
||||||
|
from src.utils.tracking import track_mcp_tool
|
||||||
|
|
||||||
|
|
||||||
|
@mcp_enabled(name="rechercher_organisations", expose_docstring=True)
|
||||||
|
def rechercher_organisations(
|
||||||
|
query: str,
|
||||||
|
type: str = "acheteur",
|
||||||
|
limite: int = 20, # noqa: A002
|
||||||
|
) -> list[dict]:
|
||||||
|
"""Recherche des acheteurs ou titulaires publics par nom.
|
||||||
|
|
||||||
|
Utiliser en premier pour résoudre un nom d'organisation vers son
|
||||||
|
identifiant, à passer ensuite à stats_acheteur / stats_titulaire.
|
||||||
|
|
||||||
|
query: texte libre (nom d'organisation).
|
||||||
|
type: "acheteur" ou "titulaire".
|
||||||
|
Retourne une liste de {id, nom, departement}.
|
||||||
|
"""
|
||||||
|
track_mcp_tool("rechercher_organisations", query=query)
|
||||||
|
return queries.search_organisations(query, type, limite)
|
||||||
|
|
||||||
|
|
||||||
|
@mcp_enabled(name="stats_acheteur", expose_docstring=True)
|
||||||
|
def stats_acheteur(acheteur_id: str) -> dict:
|
||||||
|
"""Statistiques agrégées d'un acheteur public (par identifiant).
|
||||||
|
|
||||||
|
Retourne nombre de marchés, montant total, répartition annuelle,
|
||||||
|
principaux titulaires et principaux codes CPV.
|
||||||
|
"""
|
||||||
|
track_mcp_tool("stats_acheteur")
|
||||||
|
return queries.compute_org_stats("acheteur", acheteur_id)
|
||||||
|
|
||||||
|
|
||||||
|
@mcp_enabled(name="stats_titulaire", expose_docstring=True)
|
||||||
|
def stats_titulaire(titulaire_id: str) -> dict:
|
||||||
|
"""Statistiques agrégées d'un titulaire (entreprise) par identifiant.
|
||||||
|
|
||||||
|
Retourne nombre de marchés remportés, montant total, répartition
|
||||||
|
annuelle, principaux acheteurs et principaux codes CPV.
|
||||||
|
"""
|
||||||
|
track_mcp_tool("stats_titulaire")
|
||||||
|
return queries.compute_org_stats("titulaire", titulaire_id)
|
||||||
|
|
||||||
|
|
||||||
|
@mcp_enabled(name="rechercher_marches", expose_docstring=True)
|
||||||
|
def rechercher_marches(
|
||||||
|
acheteur_id: str | None = None,
|
||||||
|
titulaire_id: str | None = None,
|
||||||
|
cpv: str | None = None,
|
||||||
|
objet_contient: str | None = None,
|
||||||
|
montant_min: float | None = None,
|
||||||
|
montant_max: float | None = None,
|
||||||
|
date_min: str | None = None,
|
||||||
|
date_max: str | None = None,
|
||||||
|
departement: str | None = None,
|
||||||
|
page: int = 1,
|
||||||
|
filtres_avances: dict | None = None,
|
||||||
|
) -> dict:
|
||||||
|
"""Recherche paginée de marchés publics (DECP).
|
||||||
|
|
||||||
|
Filtres nommés : acheteur_id, titulaire_id, cpv (code CPV, correspondance
|
||||||
|
partielle), objet_contient (texte de l'objet), montant_min, montant_max,
|
||||||
|
date_min / date_max (format YYYY-MM-DD, sur dateNotification),
|
||||||
|
departement (code département de l'acheteur).
|
||||||
|
filtres_avances : dict optionnel {"colonne__operateur": valeur} pour les
|
||||||
|
besoins pointus (mêmes colonnes/opérateurs que l'API REST colibre).
|
||||||
|
page : numéro de page (50 résultats par page).
|
||||||
|
Retourne {meta: {page, page_size, total}, marches: [...]}.
|
||||||
|
"""
|
||||||
|
track_mcp_tool("rechercher_marches", query=objet_contient)
|
||||||
|
return queries.search_marches(
|
||||||
|
acheteur_id=acheteur_id,
|
||||||
|
titulaire_id=titulaire_id,
|
||||||
|
cpv=cpv,
|
||||||
|
objet_contient=objet_contient,
|
||||||
|
montant_min=montant_min,
|
||||||
|
montant_max=montant_max,
|
||||||
|
date_min=date_min,
|
||||||
|
date_max=date_max,
|
||||||
|
departement=departement,
|
||||||
|
page=page,
|
||||||
|
filtres_avances=filtres_avances,
|
||||||
|
)
|
||||||
@@ -0,0 +1,33 @@
|
|||||||
|
from src.mcp import tools
|
||||||
|
|
||||||
|
|
||||||
|
def test_all_four_tools_are_callable():
|
||||||
|
for name in (
|
||||||
|
"rechercher_organisations",
|
||||||
|
"stats_acheteur",
|
||||||
|
"stats_titulaire",
|
||||||
|
"rechercher_marches",
|
||||||
|
):
|
||||||
|
assert callable(getattr(tools, name))
|
||||||
|
|
||||||
|
|
||||||
|
def test_rechercher_organisations_returns_list():
|
||||||
|
result = tools.rechercher_organisations("ACHETEUR", "acheteur")
|
||||||
|
assert isinstance(result, list)
|
||||||
|
assert any(r["id"] == "123" for r in result)
|
||||||
|
|
||||||
|
|
||||||
|
def test_rechercher_marches_returns_meta():
|
||||||
|
result = tools.rechercher_marches(acheteur_id="123")
|
||||||
|
assert result["meta"]["total"] >= 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_stats_acheteur_returns_stats():
|
||||||
|
result = tools.stats_acheteur("123")
|
||||||
|
assert result["nb_marches"] >= 1
|
||||||
|
assert result["identite"]["id"] == "123"
|
||||||
|
|
||||||
|
|
||||||
|
def test_stats_titulaire_returns_stats():
|
||||||
|
result = tools.stats_titulaire("345")
|
||||||
|
assert result["nb_marches"] >= 1
|
||||||
Reference in New Issue
Block a user