pip install requests

session.headers.update({
    "Authorization": "Bearer VOTRE_TOKEN_ICI"
})

url = f"https://mcp.openlegi.fr/legifrance/mcp?token=VOTRE_TOKEN"

url = "https://mcp.openlegi.fr/legifrance/mcp/token/VOTRE_TOKEN"

import requests
import json


class OpenLegiClient:
    """Client Python pour l'API OpenLegi MCP.

    Gère le protocole MCP avec support du format SSE (Server-Sent Events).
    """

    def __init__(
        self,
        token: str,
        service: str = "legifrance",
        base_url: str = "https://mcp.openlegi.fr"
    ):
        """Initialise le client MCP.

        Args:
            token: Token d'authentification OpenLegi
            service: Service à utiliser ('legifrance', 'inpi', 'eurlex')
            base_url: URL de base du serveur MCP
        """
        self.base_url = base_url
        self.service = service
        self.endpoint = f"{base_url}/{service}/mcp"
        self.session = requests.Session()

        # Headers requis par le protocole MCP
        self.session.headers.update({
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",  # Les deux types sont requis
            "Authorization": f"Bearer {token}"
        })
        # Initialiser la session MCP
        self._initialize()

    def _initialize(self):
        """Initialise la session MCP et récupère le session ID."""
        payload = {
            "jsonrpc": "2.0",
            "id": 0,
            "method": "initialize",
            "params": {
                "protocolVersion": "2024-11-05",
                "capabilities": {},
                "clientInfo": {
                    "name": "OpenLegiClient",
                    "version": "1.0.0"
                }
            }
        }

        response = self.session.post(self.endpoint, json=payload)
        response.raise_for_status()

        # Récupérer le session ID du header
        session_id = response.headers.get('mcp-session-id')
        if session_id:
            self.session.headers.update({"mcp-session-id": session_id})


    @staticmethod
    def _parse_sse_response(text: str) -> dict:
        """Parse une réponse SSE et extrait le JSON de la ligne data:.

        Le serveur MCP renvoie les réponses au format SSE :
        event: message
        data: {"jsonrpc":"2.0","id":1,"result":{...}}

        Args:
            text: Texte brut de la réponse HTTP

        Returns:
            Dictionnaire JSON extrait de la ligne 'data:'

        Raises:
            ValueError: Si aucune ligne 'data:' n'est trouvée
        """
        lines = text.strip().split('\n')
        for line in lines:
            if line.startswith('data: '):
                return json.loads(line[6:])  # Retire 'data: ' et parse le JSON
        raise ValueError("Aucune ligne 'data:' trouvée dans la réponse SSE")

    def call_tool(self, tool_name: str, arguments: dict) -> dict:
        """Appelle un tool MCP avec les arguments fournis.

        Args:
            tool_name: Nom du tool à appeler (ex: 'rechercher_code')
            arguments: Dictionnaire des arguments du tool

        Returns:
            Résultat du tool (contenu du champ 'result')

        Raises:
            requests.HTTPError: Si la requête HTTP échoue
            Exception: Si le serveur renvoie une erreur JSON-RPC
        """
        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": tool_name,
                "arguments": arguments
            }
        }

        response = self.session.post(self.endpoint, json=payload)
        response.raise_for_status()

        # Parser la réponse SSE
        result = self._parse_sse_response(response.text)

        if "error" in result:
            raise Exception(f"Erreur MCP: {result['error']}")

        return result["result"]

    def list_tools(self) -> list[dict]:
        """Liste tous les tools disponibles sur le serveur.

        Returns:
            Liste des tools avec leurs métadonnées (name, description, inputSchema)

        Raises:
            requests.HTTPError: Si la requête HTTP échoue
        """
        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/list",
            "params": {}
        }

        response = self.session.post(self.endpoint, json=payload)
        response.raise_for_status()

        # Parser la réponse SSE
        result = self._parse_sse_response(response.text)
        return result["result"]["tools"]

    def close(self):
        """Ferme la session HTTP."""
        self.session.close()

    def __enter__(self):
        """Support du context manager (with statement)."""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Fermeture automatique de la session."""
        self.close()

from openlegi_client import OpenLegiClient

# Créer le client avec votre token
token = "votre_token_ici"

with OpenLegiClient(token=token) as client:
    # Récupérer la liste des tools
    tools = client.list_tools()

    print(f"📋 {len(tools)} tools disponibles:\n")

    for tool in tools:
        print(f"🔧 {tool['name']}")
        print(f"   {tool['description'][:100]}...")
        print()

📋 12 tools disponibles:

🔧 rechercher_dans_texte_legal
   Recherche dans les textes légaux consolidés depuis légifrance...

🔧 rechercher_code
   Recherche des articles juridiques dans les codes juridiques français...

🔧 rechercher_jurisprudence_judiciaire
   Recherche des jurisprudences judiciaires avec validation MCP...

🔧 rechercher_jurisprudence_administrative
   Recherche dans la jurisprudence administrative (Conseil d'État...

🔧 rechercher_decisions_cnil
   Recherche dans les délibérations et décisions de la CNIL...

🔧 rechercher_conventions_collectives
   Recherche dans les conventions collectives (KALI)...

🔧 rechercher_decisions_constitutionnelles
   Recherche dans les décisions du Conseil constitutionnel...

🔧 recherche_journal_officiel
   Recherche dans le Journal Officiel français...

🔧 dernier_journal_officiel
   Récupère le(s) dernier(s) Journal(aux) Officiel(s) publiés...

🔧 lister_codes_juridiques
   Liste tous les codes juridiques disponibles sur Legifrance...

🔧 lister_emetteurs_jorf
   Liste tous les émetteurs/autorités disponibles pour les recherches JORF...

🔧 lister_natures_textes_jorf
   Liste toutes les natures de textes disponibles pour les recherches JORF...

from openlegi_client import OpenLegiClient

token = "votre_token_ici"

with OpenLegiClient(token=token) as client:
    # Rechercher l'article 9 du Code civil
    result = client.call_tool(
        tool_name="rechercher_code",
        arguments={
            "search": "9",
            "code_name": "Code civil",
            "champ": "NUM_ARTICLE"
        }
    )

    print("📖 Résultat de la recherche:\n")
    print(result)

from openlegi_client import OpenLegiClient

token = "votre_token_ici"

with OpenLegiClient(token=token) as client:
    # Rechercher des décisions sur la responsabilité civile
    result = client.call_tool(
        tool_name="rechercher_jurisprudence_judiciaire",
        arguments={
            "search": "responsabilité civile",
            "juridiction_judiciaire": ["Cour de cassation"],
            "publication_bulletin": ["T"],
            "sort": "DATE_DESC",
            "page_size": 5
        }
    )

    print("⚖️ Jurisprudences trouvées:\n")
    print(result)

from openlegi_client import OpenLegiClient

token = "votre_token_ici"

with OpenLegiClient(token=token) as client:
    # Récupérer les 3 derniers JO
    result = client.call_tool(
        tool_name="dernier_journal_officiel",
        arguments={
            "nb_jo": 3
        }
    )

    print("📰 Derniers Journaux Officiels:\n")
    print(result)

from openlegi_client import OpenLegiClient

token = "votre_token_ici"

with OpenLegiClient(token=token) as client:
    # Rechercher des conventions sur le télétravail
    result = client.call_tool(
        tool_name="rechercher_conventions_collectives",
        arguments={
            "search": "télétravail",
            "panorama": True,  # Vue d'ensemble sans texte intégral
            "page_size": 10
        }
    )

    print("📄 Conventions collectives:\n")
    print(result)

from openlegi_client import OpenLegiClient

token = "votre_token_ici"

with OpenLegiClient(token=token) as client:
    # Lister tous les codes disponibles
    result = client.call_tool(
        tool_name="lister_codes_juridiques",
        arguments={}
    )

    print("📚 Codes juridiques disponibles:\n")
    for code in result:
        print(f"  • {code}")

from openlegi_client import OpenLegiClient
import requests

token = "votre_token_ici"

try:
    with OpenLegiClient(token=token) as client:
        result = client.call_tool("rechercher_code", {"search": "test"})
except requests.HTTPError as e:
    print(f"❌ Erreur HTTP: {e}")
    print(f"Status: {e.response.status_code}")
    print(f"Détails: {e.response.text}")

from openlegi_client import OpenLegiClient

token = "votre_token_ici"

try:
    with OpenLegiClient(token=token) as client:
        # Tool inexistant
        result = client.call_tool("tool_qui_nexiste_pas", {})
except Exception as e:
    print(f"❌ Erreur MCP: {e}")

event: message
data: {"jsonrpc":"2.0","id":1,"result":{...}}