IdentifiantMot de passe
Loading...
Mot de passe oublié ?Je m'inscris ! (gratuit)
Navigation

Inscrivez-vous gratuitement
pour pouvoir participer, suivre les réponses en temps réel, voter pour les messages, poser vos propres questions et recevoir la newsletter

Contribuez Python Discussion :

MCP - La Nouvelle Hype


Sujet :

Contribuez Python

Vue hybride

Message précédent Message précédent   Message suivant Message suivant
  1. #1
    Expert confirmé Avatar de papajoker
    Homme Profil pro
    Développeur Web
    Inscrit en
    Septembre 2013
    Messages
    2 291
    Détails du profil
    Informations personnelles :
    Sexe : Homme
    Localisation : France, Nièvre (Bourgogne)

    Informations professionnelles :
    Activité : Développeur Web
    Secteur : High Tech - Multimédia et Internet

    Informations forums :
    Inscription : Septembre 2013
    Messages : 2 291
    Par défaut MCP - La Nouvelle Hype
    Bonjour,

    Découvrez MCP : La Nouvelle Hype pour les Développeurs IA !

    Vous ne connaissez pas encore MCP ? C'est pourtant le sujet d'enthousiasme du moment pour les développeurs ayant un intérêt pour l'intelligence artificielle.
    **Attention, nous n'en sommes qu'aux balbutiements de cette technologie.

    Imaginez un monde où vos scripts et applications communiquent nativement avec n'importe quelle IA. Ce futur est plus proche que vous ne le pensez grâce à MCP.
    Dès demain, allons-nous créer des serveurs MCP pour héberger tous nos scripts et applications ? Pourquoi cet engouement ?
    - Demande de nos clients.
    - Simplicité déconcertante pour ajouter cette fonctionnalité à notre travail.
    - Le standard de demain avec l'IA ?


    Si vous viviez dans une grotte...
    MCP (Middleware Communication Protocol) s’impose comme une couche d’abstraction universelle entre les environnements d’exécution applicatifs et les moteurs d’IA.
    En pratique, un développeur peut, en quelques lignes de code, invoquer une fonction IA distante, recevoir une réponse contextualisée, et chaîner cette réponse avec d’autres modules métiers sans se soucier de la compatibilité technique entre les composants. MCP fait le lien entre les paradigmes de l’IA et ceux du développement logiciel traditionnel, en supprimant la friction entre les deux mondes.

    Pour nous, développeurs (et non administrateurs), ce protocole se révèle très simple à appréhender. Il nous suffit de créer une interface très proche de REST. Deux modes de communication existent : `io` pour une communication locale, et `http` pour une communication réseau, donc publique (avec authentification possible).

    Une librairie Python haut niveau "fastmcp" existe (qui utilise mcp) pour nous simplifier grandement la vie. L'exportation d'une fonction se fait de la manière suivante :
    Code : Sélectionner tout - Visualiser dans une fenêtre à part
    1
    2
    3
    4
    5
    6
    @mcp.tool()
    def ma_fonction_exportée(
        param: str = Field(description="descriptif pour IA"),
    ) :
        """texte d'explication pour l'IA sur cette fonction"""
        return "ok"
    Rien de plus simple, n'est-ce pas ? Il est crucial de comprendre que les annotations sont primordiales, car elles seront utilisées par l'IA pour déterminer si notre fonction est pertinente pour la requête de l'utilisateur.

    Cas d'usages simples :
    Lecture :
    - Vous avez déjà un script qui retourne des informations ? Créez un serveur MCP qui l'inclura et offrira ces mêmes informations à toutes les IA.
    - Votre application peut retourner la liste de vos clients, le stock, etc. Exposez ces données via MCP !
    - Vous possédez une API (REST ou autre) ? MCP peut la consulter ou l'étendre.
    Action :
    - En plus, piloter un périphérique
    - Le principe est similaire à la lecture pour nous (mais à voir si l'ia a bien compris le prompt... ).



    Explorez cette belle liste de serveurs MCP
    Chaque jour il y a de nouvelles entrées dans la catégorie mcp sur dev.to


    Notez que certaines IA peuvent enchaîner les requêtes MCP. Par exemple, consulter des données provenant d'un serveur MCP, puis, en fonction du résultat, solliciter une "action" auprès d'un autre serveur MCP.

    "Communication entre "toutes*" les IA" :
    Attention ! Ceci est un horizon futur. Nous n'en sommes qu'aux prémices. Toutes les IA ne supportent pas encore ce protocole, et surtout : chaque IA n'interprétera pas forcément la requête utilisateur de la même manière. Le risque est donc de voir l'IA appeler une fonction inappropriée ou avec les mauvais paramètres. De plus, la possibilité de connecter une centaine de serveurs à notre IA introduit un risque de confusion et de collision.


    Il est temps de présenter un petit code fonctionnel.

    `fastmcp` a été utilisé ici, une surcouche de haut niveau qui simplifie notre travail.

    Code : Sélectionner tout - Visualiser dans une fenêtre à part
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    #!/usr/bin/env -S uv run --script
    # /// script
    # requires-python = "<=3.14"
    # dependencies = ["fastmcp", "beautifulsoup4"]
    # ///
     
    """
    dans ia, test de quelques prompts:
      les derniers messages du forum
      les derniers messages du forum dans la catégorie administration
    """
     
    import asyncio
    import urllib.request
    import sys
    from bs4 import BeautifulSoup
     
    from pydantic import Field
    from mcp.server.fastmcp import FastMCP
     
     
    mcp = FastMCP(
        "serveur-mcp-developpez",
        log_level="ERROR",  # pouvons changer pour DEBUG
        instructions="""
            informations sur le forum `developpez`
            récupérer les derniers sujets, messages du forum `developpez`
            return : json format
            json type is:
                success : yes or not
                title : response title
                data : response datas
                error : if error, text
        """,
    )
     
     
    def get_http_page():
        """
        fonction métier, venant de motre application
        from mon_application import get_http_page
        """
        URL = "https://www.developpez.net/forums/search.php?do=getdaily"
        html = urllib.request.urlopen(URL).read().decode("ISO-8859-1")
     
        # code rapide pour ne garder que le texte
        soup = BeautifulSoup(html, "html.parser")
        results = []
        for item in soup.select("li.imodselector"):
            try:
                title = item.select("h3")[0].get_text().strip()
                url = item.select("h3 a")[0]["href"]
                cat = item.select("div.threadpostedin a")[0].get_text().strip()
                results.append({"title": title, "cat": cat, "url": url})
            except Exception:
                continue
        return results
     
     
    class ToolResult(dict):
        """facultatif, format structuré de tous nos retours de notre api"""
     
        def __init__(self, success, title="", data=None, error=None):
            self["success"] = success
            self["title"] = title
            self["data"] = data if data else None
            # if not self["data"] : self["success"] = False; ...
            self["error"] = error
     
     
    @mcp.tool()
    async def get_messages_forum(
        category: str = Field(description="en optionnel, filtrer une catégorie", default="all"),
    ) -> dict:
        """
        return the content of the `developpez.net` forum
        return type is json
        category == `cat` in json
        Le contenu est en francais
        Dans la liste des messages, il peut avoir une duplication des sujets
        """
     
        # category : juste démo, non utilisé ici, IA peut filtrer la réponse complete (mais pas bon pour jetons)
        try:
            html = get_http_page()
            if html:
                return ToolResult(True, "messages list developpez.net", html)
            else:
                return ToolResult(False, "forum error", None, "no message found")
     
        # except ConnectionRefusedError:
        # return {"status": 0, "message": f"Erreur: Impossible de se connecter"}
        except Exception as e:
            return ToolResult(False, "forum error", None, f"read forum error : {e}")
     
     
    async def main():
        """
        juste pour un debug simple,
        normalement fonctions sont déjà présentent et testées dans notre application
        """
        print(await get_messages_forum())
        exit(0)
     
     
    if __name__ == "__main__":
        if "-t" in sys.argv:
            asyncio.run(main())
        mcp.run(transport="stdio")  # uniquement en mode local
    Notez que l'utilisation de `uv` ici est arbitraire. Pour un déploiement plus simple, la PEP 723 n'est pas exclusive à `uv`, et il est possible de s'en passer.

    ------------------------------------

    Pour tester notre serveur, toute IA supportant MCP fera l'affaire ! Par exemple en hôtes existe: VS Code avec Copilot (doc) ou AnythingLLM Desktop.
    ** vscode-copilot avec gemini : bug avec mcp mais fixé dans la prochaine release

    Gemini a annoncé son support... mais il est trop tôt pour émettre un jugement sur cet embryon. Pour l'instant, c'est très léger, mais suffisant pour tester. Et puisqu'il est possible d'obtenir une clé API gratuitement, autant en profiter.

    Nous pouvons créer un petit client pour tester notre serveur.
    Deux cas de figure se présentent : soit nous effectuons une action, et la réponse a peu d'importance, soit nous demandons des informations. Dans ce dernier cas, la documentation nous demande d'injecter le résultat de la requête MCP dans la requête à l'IA. Soyez vigilants, les jetons en entrée correspondent à la sortie de notre fonction.

    Voici un petit client Gemini connecté à notre serveur qui interprète la réponse de notre serveur MCP.

    Code : Sélectionner tout - Visualiser dans une fenêtre à part
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    #!/usr/bin/env -S uv run --script
    # /// script
    # requires-python = "<=3.13.20"
    # dependencies = ["mcp", "google>=3.0.0","google-genai>=1.7.0"]
    # ///
     
    """
    https://ai.google.dev/gemini-api/docs/function-calling?hl=fr&example=weather#use_model_context_protocol_mcp
    """
     
    import asyncio
    from pathlib import Path
    import os
     
    from google import genai
    from google.genai import types
    from mcp import ClientSession, StdioServerParameters
    from mcp.client.stdio import stdio_client
     
    MCP_SERVER = Path(__file__).parent / "mcp-developpez.py"
    client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
     
     
    def set_params(server_file=MCP_SERVER):
        # Create server parameters for stdio connection
        print("server at:", server_file)
        return StdioServerParameters(
            command=str(server_file),  # Executable
            args=[],
            # command="python", args=[server_file],
            # command="uv", args=["run", "c:/xxxx/mcp-developpez.py"],
            # env=os.environ,
            env=None,
        )
     
     
    async def run(server_params):
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                # Get tools from MCP session and convert to Gemini Tool objects
                mcp_tools = await session.list_tools()
                tools = [
                    types.Tool(
                        function_declarations=[
                            {
                                "name": tool.name,
                                "description": tool.description,
                                "parameters": {
                                    k: v for k, v in tool.inputSchema.items() if k not in ["additionalProperties", "$schema"]
                                },
                            }
                        ]
                    )
                    for tool in mcp_tools.tools
                ]
     
                while True:
                    # Demande de l'utilisateur
                    print("# exemple: les derniers messages du forum")
                    prompt = input("\n(IA-CLIENTE) Ask : ")
                    if not prompt:
                        break
     
                    # Send request to the model with MCP function declarations
                    response = client.models.generate_content(
                        model="gemini-2.0-flash",
                        contents=prompt,
                        config=types.GenerateContentConfig(
                            temperature=0.5,
                            tools=tools,
                            # system_instruction=f"use my lang : {os.getenv("LANG")}",
                        ),
                    )
     
                    # print("datas brutes du serveur mcp...")
                    # print(response)
                    print(
                        "#",
                        response.usage_metadata.total_token_count,
                        "jetons utilisés pour interpréter mon prompt et trouver la fonction MCP",
                    )
                    print()
     
                    if response.candidates[0].content.parts[0].function_call:
                        function_call = response.candidates[0].content.parts[0].function_call
     
                        print("# appel de cette fonction dans le serveur MCP:", function_call)
                        # on injecte les datas du mcp dans le prompt utilisateur
                        result = await session.call_tool(
                            function_call.name,
                            arguments=function_call.args,
                        )
     
                        # ici, nous avons le retour brut de la requete à MCP
                        # print("\n".join(r.text for r in result.content))
     
                        # eventuellement , encore une requete a ia pour mettre en clair
                        # on passe le résultat MCP dans le contexte ...
                        # https://ai.google.dev/gemini-api/docs/function-calling?hl=fr&example=weather#step_4_create_user_friendly_response_with_function_result_and_call_the_model_again
     
                        contents = [
                            types.Content(
                                role="user",
                                parts=[types.Part(text="not display json input., but format json result to humain text")],
                            )
                        ]
     
                        # Create a function response part
                        function_response_part = types.Part.from_function_response(
                            name=function_call.name,
                            response={"result": result.content},
                        )
     
                        # Append function call and result of the function execution to contents
                        contents.append(
                            types.Content(role="model", parts=[types.Part(function_call=function_call)])
                        )  # Append the model's function call message
                        contents.append(types.Content(role="user", parts=[function_response_part]))  # Append the function response
     
                        final_response = client.models.generate_content(
                            model="gemini-2.0-flash",
                            config=types.GenerateContentConfig(
                                temperature=0.5,
                                tools=tools,
                                system_instruction=f"""
                                    use my lang : {os.getenv("LANG")}
                                """,
                            ),
                            contents=contents,
                        )
     
                        print(final_response.text)
     
                        print()
                        print("#", final_response.usage_metadata.total_token_count, "jetons utilisés encore en plus")
                        print()
     
                    else:
                        print("Pas d'utilisation du serveur MCP pour la réponse !")
                        print(response.text)
     
     
    if __name__ == "__main__":
        serv = set_params(MCP_SERVER)
        asyncio.run(run(serv))
    Résultat du mini client, il a trouvé (actuellement) 2 sujets:
    Code : Sélectionner tout - Visualiser dans une fenêtre à part
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    (IA-CLIENTE) Ask : les derniers messages du forum dans la catégorie administration
    # 87 jetons utilisés
     
    # appel de cette fonction dans le serveur MCP: id=None args={'category': 'administration'} name='get_messages_forum'
    Voici le contenu du forum developpez.net, catégorie "administration" :
     
    *   **Comment créer une vraie copie d'une bdd en ssh ?** dans Administration: <https://www.developpez.net/forums/d2175945/bases-donnees/mysql/administration/creer-vraie-copie-d-bdd-ssh/>
    *   **[2017] Quelle limitation taille base SQL Express 2017 ?** dans Administration: <https://www.developpez.net/forums/d2176398/bases-donnees/ms-sql-server/administration/limitation-taille-base-sql-express-2017-a/>
     
    # 2528 jetons utilisés encore en plus
    Code : Sélectionner tout - Visualiser dans une fenêtre à part
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    (IA-CLIENTE) Ask : les derniers messages du forum
    # 81 jetons utilisés
     
    # appel de cette fonction dans le serveur MCP: id=None args={} name='get_messages_forum'
    Voici les sujets du forum developpez.net :
     
    *   **Qt Creator:** Comment redimensionner les éléments d'une appli
    *   **Sécurité:** Les gouvernements ont plus que jamais recours à des piratages de type "zero-day"
    *   **CV:** CV pour emploi et/ou stage
    *   **Firefox:** Nouvelle version Waterfox v6.5.7
    *   **Schéma:** \[MCD] Inventaire livres bibliothèque personnelle
    *   **WinDev:** \[WD23] Perte de l'aide en ligne
    *   **Linux:** Linus Torvalds exprime son mécontentement à l'égard des systèmes de fichiers insensibles à la casse
    *   **Hardware:** Meta transforme ses lunettes connectées Ray-Ban en machine de surveillance pour l'IA
    *   **Actualités:** Un développeur signale que Microsoft GitHub Copilot s'est activé sans son consentement explicite
    *   **Sécurité:** Certaines applications prennent automatiquement des captures d'écran et les envoient à des tiers
    *   **Débuter:** problème d'accent avec wamp
    *   **Intelligence artificielle:** Wikipédia a déclaré qu'elle utiliserait l'IA pour améliorer le travail de ses rédacteurs et de ses bénévoles
    *   **Débuter:** Utilisation du TopenDialog au lieu du nom de dossier avec D6 et Win11 64bits
    *   **Extensions:** \[PostGIS] Charger des primitives différentes
    *   **Administration:** Comment créer une vraie copie d'une bdd en ssh ?
    *   **Administration:** \[2017] Quelle limitation taille base SQL Express 2017 ?
    *   **Macros et VBA Excel:** automatiser copier coller en VBE
    *   **jQuery:** Une petite question sur un element presque fini
    *   **Langage SQL:** Sélectionner les éléments d'une table qui ne sont pas dans une autre
    *   **React:** Connexion base de données - récupérer les messages d'erreur
    *   **Outlook:** \[Toutes versions] Outlook & VBA / Selection d'un répertoire ou d'un fichier
    *   **C#:** \[Débutant] Remplacer des strings dans une liste
    *   **Outlook:** \[OL-365] Copier les rendez-vous d'un calendrier partagé vers un nouveau calendrier
    *   **Outlook:** \[OL-2019] Outlook 2021 - Accusé de réception ne fonctionne plus
    *   **Macros et VBA Excel:** \[Toutes versions] Selection de répertoire et de fichiers
     
    # 2946 jetons utilisés encore en plus
    ----------

    A noter que fastmcp n'est pas la seule solution python, existe aussi fastapi-mcp qui expose notre serveur mcp comme un serveur http
    Ces 2 librairies utilisent mcp "The official Python SDK for Model Context Protocol servers and clients"

    FastAPI MCP est une solution élégante qui vous permet d'exposer vos points de terminaison FastAPI en tant qu'outils MCP (Model Context Protocol) avec un minimum d'effort.
    Code : Sélectionner tout - Visualiser dans une fenêtre à part
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    #src: https://github.com/tadata-org/fastapi_mcp/blob/main/examples/02_full_schema_description_example.py
     
    @app.get("/items/{item_id}", response_model=Item, tags=["items"], operation_id="get_item")
    async def read_item(item_id: int):
        """
        Get a specific item by its ID.
        Raises a 404 error if the item does not exist.
        """
        if item_id not in items_db:
            raise HTTPException(status_code=404, detail="Item not found")
        return items_db[item_id]
     
    if __name__ == "__main__":
        import uvicorn
        uvicorn.run(app, host="127.0.0.1", port=PORT)
    Pour faire le lien avec fastmcp, operation_id est en fait le nom de la fonction (tool) mcp transmise à l'ia


    ----------

    À suivre... L'IA étant pilotable par la voix (la voix générant simplement un prompt texte), verrons-nous demain une IA intégrée à notre OS, où tout sera contrôlable vocalement ?
    $moi= (:nono: !== :oops:) ? :king: : :triste: ;

  2. #2
    Membre chevronné
    Homme Profil pro
    web a11y
    Inscrit en
    Avril 2014
    Messages
    188
    Détails du profil
    Informations personnelles :
    Sexe : Homme
    Localisation : France, Loire Atlantique (Pays de la Loire)

    Informations professionnelles :
    Activité : web a11y
    Secteur : Service public

    Informations forums :
    Inscription : Avril 2014
    Messages : 188
    Par défaut
    Certain de ne pas avoir confondu le 01/05 avec le 01/04 ?
    Nan ?
    Ah miiiince…

  3. #3
    Expert confirmé Avatar de papajoker
    Homme Profil pro
    Développeur Web
    Inscrit en
    Septembre 2013
    Messages
    2 291
    Détails du profil
    Informations personnelles :
    Sexe : Homme
    Localisation : France, Nièvre (Bourgogne)

    Informations professionnelles :
    Activité : Développeur Web
    Secteur : High Tech - Multimédia et Internet

    Informations forums :
    Inscription : Septembre 2013
    Messages : 2 291
    Par défaut
    L'IA étant pilotable par la voix (la voix générant simplement un prompt texte), verrons-nous demain une IA intégrée à notre OS, où tout sera contrôlable vocalement ?
    Je suis un rêveur, autant essayer de créer une application graphique pilotée par la voie (ou prompt texte classique).



    On commence par la fin, résultats dans un client maison:
    Code : Sélectionner tout - Visualiser dans une fenêtre à part
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    (TK-APP) Ask : envoyer bonjour
     
    # 260 jetons utilisés
     
    Pas d'utilisation du serveur MCP pour la réponse !
    Pour envoyer "bonjour", j'ai besoin de savoir à quelle application je dois l'envoyer. Est-ce que tu veux que je l'envoie à TK-APP ? Si oui, je peux utiliser la fonction `send_info`.
     
    (TK-APP) Ask : envoyer bonjour a TK-APP
     
    # 226 jetons utilisés
    # function call à envoyer à MCP: id=None args={'prompt': 'envoyer bonjour a TK-APP', 'message': 'bonjour'} name='send_info'
    {"success": false, "title": "Erreur lors de l'envoi de la commande MESSAGE \u00e0 `TK-APP`.", "data": "Erreur: Impossible de se connecter \u00e0 `TK-APP` sur localhost:12346", "error": "Impossible de contacter `TK-APP`."}
     
    (TK-APP) Ask : envoi bonjour a TK-APP 
     
    # 227 jetons utilisés
    # function call à envoyer à MCP: id=None args={'message': 'bonjour', 'prompt': 'envoi bonjour a TK-APP'} name='send_info'
    {"success": true, "title": "Commande 'message' envoy\u00e9e \u00e0 `TK-APP` avec le message: `bonjour`", "data": "bonjour", "error": null}
    L'application graphique n'est lancée qu'au dernier essai !


    mcp-app.py :
    Code : Sélectionner tout - Visualiser dans une fenêtre à part
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180
    181
    182
    183
    184
    185
    186
    187
    #!/usr/bin/env -S uv run --script
    # /// script
    # requires-python = "<=3.14"
    # dependencies = ["fastmcp"]
    # ///
     
    """
    dans ia, test de quelques prompts:
      envoyer un bonjour a mon application TK-APP Ok
      envoyer un bonjour a mon application  #BUG ? NON refusé, alors que température est ok ...
      fermer TK-APP
      sauver le fichier : OK refusé
      envoyer un "bonjour" : OK refusé
      quelle est la température : OK refusé
      quelle est la température de mon application : OK refusé
      quelle est la température de TK-APP : ok
      température de TK-APP : ok
    """
     
    import json
    import socket
     
    from pydantic import Field
    from mcp.server.fastmcp import FastMCP
     
     
    import app  # les constantes de mon application
     
     
    mcp = FastMCP(
        "serveur-mcp-{app.APP_NAME}",
        log_level="ERROR",  # can change for DEBUG
        instructions=f"""
        pilote {app.APP_NAME}.
        
        If `{app.APP_NAME}` is not explicitly in the prompt, do not use these functions.
        use only if `{app.APP_NAME}` is in prompt !
     
        return json structure as:
          - `success` : boolean
          - `title` : resume return
          - `error` : message if `success` is false
          - `data` : datas return by function
     
        """,
    )
     
     
    class ToolResult(dict):
        """format structuré de tous nos retours de notre api"""
     
        def __init__(self, success, title="", data=None, error=None):
            self["success"] = success
            self["title"] = title
            self["data"] = data if data else None
            self["error"] = error
     
     
    async def send_command_to_tk(command: str, data: dict = None):
        """Envoie une commande JSON à app graphique via un socket TCP."""
        # retourne status de l'app tk
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
                sock.connect((app.HOST, app.PORT))
                message = {"command": command, "data": data}
                sock.sendall(json.dumps(message).encode("utf-8"))
     
                response_bytes = sock.recv(1024)
                if response_bytes:
                    return json.loads(response_bytes.decode("utf-8"))
                else:
                    return {"status": 0, "message": "Aucune réponse reçue de l'application"}
     
        except ConnectionRefusedError:
            return {"status": 0, "message": f"Erreur: Impossible de se connecter à `{app.APP_NAME}` sur {app.HOST}:{app.PORT}"}
        except Exception as e:
            return {"status": 0, "message": f"Erreur lors de l'envoi de la commande à `{app.APP_NAME}`: {e}"}
     
     
    @mcp.tool()
    async def send_info(
        message,  #: str = Field(description="Le message à envoyer et afficher"),
        prompt: str = Field(description="user prompt (interne à fastmcp)"),
    ) -> ToolResult:
        """
        Envoyer un message texte à TK-APP qui va afficher ce message.
        use only if TK-APP is in prompt !
        """
     
        """ Pas utile, si "prompt" est forcé dans les params (avec gemini 2.0 ...)
        if APP_NAME not in prompt_.lower(): return Error
        """
     
        response = await send_command_to_tk("send_info", {"message": message})
     
        if not response.get("status"):
            return ToolResult(
                False,
                f"Erreur lors de l'envoi de la commande MESSAGE à `{app.APP_NAME}`.",
                response.get("message"),
                f"Impossible de contacter `{app.APP_NAME}`.",
            )
     
        return ToolResult(
            True,
            f"Commande 'message' envoyée à `{app.APP_NAME}` avec le message: `{message}`",
            response.get("message"),
        )
     
     
    @mcp.tool()
    async def save_file(
        prompt: str = Field(description="user prompt (interne à fastmcp)"),
    ) -> ToolResult:
        """
        Enregister le fichier courant ouvert par l'application.
        use only if word TK-APP is in prompt !
        """
        response = await send_command_to_tk("save", {"message": ""})
     
        if not response.get("status"):
            return ToolResult(
                False,
                f"Erreur lors de l'envoi de la commande à `{app.APP_NAME}`.",
                response.get("message"),
                f"Impossible de contacter `{app.APP_NAME}`.",
            )
     
        return ToolResult(
            True,
            f"Commande bien exécutée par `{app.APP_NAME}`",
            "toto.txt",
        )
     
     
    @mcp.tool()
    async def close_app(
        prompt: str = Field(description="user prompt (interne à fastmcp)"),
    ) -> ToolResult:
        """
        Close and exit application
        use only if word TK-APP is in prompt !
        """
        response = await send_command_to_tk("exit", None)
     
        if not response.get("status"):
            return ToolResult(
                False,
                f"Erreur lors de l'envoi de la commande QUITTER à `{app.APP_NAME}`.",
                response.get("message"),
                f"Impossible de contacter `{app.APP_NAME}`.",
            )
     
        return ToolResult(
            True,
            f"Commande QUITTER bien exécutée par `{app.APP_NAME}`",
        )
     
     
    @mcp.tool()
    async def temp_app(
        prompt: str = Field(description="user prompt (interne à fastmcp)"),
    ) -> ToolResult:
        """
        connaitre la température de l'application
        use only if word TK-APP is in prompt !
        """
     
        response = await send_command_to_tk("fake", {"message": ""})
     
        if not response.get("status"):
            return ToolResult(
                False,
                f"Erreur lors de l'envoi de la commande `température` à `{app.APP_NAME}`.",
                response.get("message"),
                f"Impossible de contacter `{app.APP_NAME}`.",
            )
     
        return ToolResult(
            True,
            "température",
            response.get("message"),
        )
     
     
    if __name__ == "__main__":
        mcp.run(transport="stdio")  # uniquement en mode local
    Ici, la seule particularité est de passer le prompt en paramètre. Pourquoi ? c'est ma seule solution avec gemini pour restreindre la fonction à un mot clé.
    Avec une fonction comme fermer/quitter , c'est la moindre des choses


    app.py :
    application tk, avec une connection au serveur mcp via socket
    PS : Pour l'application graphique (TK-APP), j'ai demandé à une IA de me l'écrire, comme le protocole socket.
    Code : Sélectionner tout - Visualiser dans une fenêtre à part
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    #!/usr/bin/env python
    import tkinter as tk
    import socket
    import threading
    import json
    import tkinter.messagebox
     
    APP_NAME = "TK-APP"
    HOST = "localhost"
    PORT = 12346
     
     
    class TkApp:
        def __init__(self, master):
            self.master = master
            master.title("Application Tkinter Pilotée par ia*")
     
            self.label = tk.Label(master, text="commandes via IA...")
            self.label.pack(pady=10)
     
            self.log_text = tk.Text(master, height=10, width=40)
            self.log_text.pack(padx=10, pady=10)
            self.log_text.config(state=tk.DISABLED)
     
            self.stop_flag = threading.Event()
            self.server_thread = threading.Thread(target=self.run_server)
            self.server_thread.daemon = True
            self.server_thread.start()
     
        def log(self, message):
            self.log_text.config(state=tk.NORMAL)
            self.log_text.insert(tk.END, message + "\n")
            self.log_text.config(state=tk.DISABLED)
            self.log_text.see(tk.END)
     
        def handle_send_info(self, data, client_socket):
            """Affiche un message via une boîte de dialogue."""
            message = data.get("message", "Message non spécifié ?")
            self.log(f"Commande 'send_info' reçue avec le message: {message}")
     
            response_data = {"status": 1, "message": message}
            client_socket.sendall(json.dumps(response_data).encode("utf-8"))
     
            tkinter.messagebox.showinfo("Bouton Poussé!", f"Message du serveur: {message}")
     
        def handle_fake(self, data, client_socket):
            # lire dans DB et retourner résultat à l'ia
            response_data = {
                "status": 1,
                "message": [
                    ["janvier", "février", "mars", "avril", "mai"],
                    [548, 799, 5, 12, 788],
                ],
            }
            client_socket.sendall(json.dumps(response_data).encode("utf-8"))
     
        def manage_ia_commands(self, command, data, client_socket):
            match command:
                case "send_info":
                    self.handle_send_info(data, client_socket)
                case "exit":
                    self.on_closing()
                    exit(0)
                case "fake":
                    self.handle_fake(data, client_socket)
                case _:
                    self.log(f"?? commande ia inconnue: {command}")
                    response_data = {"status": 0, "message": "commande ia inconnue"}
                    client_socket.sendall(json.dumps(response_data).encode("utf-8"))
     
        def handle_client(self, client_socket, client_address):
            # TODO répondre à MCP si commande est ok
            self.log(f"Serveur Tkinter: Client connecté: {client_address}")
            try:
                while not self.stop_flag.is_set():
                    data = client_socket.recv(1024)
                    if not data:
                        break
                    try:
                        request = json.loads(data.decode("utf-8"))
                        command = request.get("command")
                        datas = request.get("data", {})
                        self.log(f"Serveur Tkinter: Commande reçue: {command}, Données: {datas}")
     
                        self.manage_ia_commands(command, datas, client_socket)
     
                    except json.JSONDecodeError as e:
                        self.log("Serveur Tkinter: Erreur de décodage JSON.")
                        response_data = {"status": 0, "message": str(e)}
                        client_socket.sendall(json.dumps(response_data).encode("utf-8"))
                    except Exception as e:
                        self.log(f"Serveur Tkinter: Erreur lors du traitement de la requête: {e}")
                        response_data = {"status": 0, "message": str(e)}
                        client_socket.sendall(json.dumps(response_data).encode("utf-8"))
     
            except ConnectionResetError:
                self.log(f"Serveur Tkinter: Client déconnecté de force: {client_address}")
            except Exception as e:
                self.log(f"Serveur Tkinter: Erreur dans la communication avec le client {client_address}: {e}")
            finally:
                client_socket.close()
                self.log(f"Serveur Tkinter: Connexion avec le client {client_address} fermée.")
     
        def run_server(self):
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
                try:
                    server_socket.bind((HOST, PORT))
                    server_socket.listen()
                    self.log(f"Serveur Tkinter démarré sur {HOST}:{PORT}")
                    while not self.stop_flag.is_set():
                        client_socket, client_address = server_socket.accept()
                        client_thread = threading.Thread(target=self.handle_client, args=(client_socket, client_address))
                        client_thread.daemon = True
                        client_thread.start()
                except OSError as e:
                    self.log(f"Serveur Tkinter: Erreur lors du démarrage du serveur: {e}")
                finally:
                    self.log("Serveur Tkinter arrêté.")
     
        def on_closing(self):
            self.log("Serveur Tkinter: Arrêt du serveur...")
            self.stop_flag.set()
            self.server_thread.join(timeout=1)
            self.master.destroy()
     
     
    if __name__ == "__main__":
        root = tk.Tk()
        app = TkApp(root)
        root.protocol("WM_DELETE_WINDOW", app.on_closing)
        root.mainloop()
    Le petit client :
    Code : Sélectionner tout - Visualiser dans une fenêtre à part
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    #!/usr/bin/env -S uv run --script
    # /// script
    # requires-python = "<=3.14"
    # dependencies = ["mcp", "google>=3.0.0","google-genai>=1.7.0"]
    # ///
     
    """
    https://ai.google.dev/gemini-api/docs/function-calling?hl=fr&example=weather#use_model_context_protocol_mcp
    """
     
    import asyncio
    import json
    from pathlib import Path
    import os
     
    from google import genai
    from google.genai import types
    from mcp import ClientSession, StdioServerParameters
    from mcp.client.stdio import stdio_client
     
    MCP_SERVER = Path(__file__).parent / "mcp-app.py"
    client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
     
     
    def set_params(server_file=MCP_SERVER):
        # Create server parameters for stdio connection
        print("server at:", server_file)
        return StdioServerParameters(
            command=str(server_file),  # Executable
            args=[],
            # command="python", args=["server_file"],
            # env=os.environ,
            env=None,
        )
     
     
    async def run(server_params):
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                # Get tools from MCP session and convert to Gemini Tool objects
                mcp_tools = await session.list_tools()
                tools = [
                    types.Tool(
                        function_declarations=[
                            {
                                "name": tool.name,
                                "description": tool.description,
                                "parameters": {
                                    k: v for k, v in tool.inputSchema.items() if k not in ["additionalProperties", "$schema"]
                                },
                            }
                        ]
                    )
                    for tool in mcp_tools.tools
                ]
     
                while True:
                    # Demande de l'utilisateur
                    prompt = input("\n(TK-APP) Ask : ")
                    if not prompt:
                        break
     
                    # Send request to the model with MCP function declarations
                    response = client.models.generate_content(
                        model="gemini-2.0-flash",
                        contents=prompt,
                        config=types.GenerateContentConfig(
                            temperature=1,
                            tools=tools,
                            system_instruction=f"""
                                use my lang : {os.getenv("LANG", "fr")}
                                after function call, format json returned by tool function to humain text.
                                """,
                        ),
                    )
     
                    print()
     
                    """# pour voir requete et détail des jetons consommés
                    print(response)"""
                    print("#", response.usage_metadata.total_token_count, "jetons utilisés")
                    print()
     
                    if response.candidates[0].content.parts[0].function_call:
                        function_call = response.candidates[0].content.parts[0].function_call
     
                        print("# function call à envoyer à MCP:", function_call)
                        result = await session.call_tool(
                            function_call.name,
                            arguments=function_call.args,
                        )
                        print("\n".join(r.text for r in result.content))
     
                    else:
                        print("Pas d'utilisation du serveur MCP pour la réponse !")
                        print(response.text)
     
     
    if __name__ == "__main__":
        serv = set_params(MCP_SERVER)
        asyncio.run(run(serv))

    Voilà, je pilote mon application graphique via IA, maintenant, je suis en attendre de voir cela un jour en production.
    Ce rêve n'est pas si stupide et relativement simple à mettre en œuvre dès aujourd'hui.
    Pour le voie, pas dans notre client ! Passer par VS Code avec Copilot ou AnythingLLM Desktop ou autres.


    Note: je suis sous kde plasma, je pourrai facilement créer un serveur MCP pour piloter mon OS ...
    Beaucoup de commandes sont disponibles via dbus
    $moi= (:nono: !== :oops:) ? :king: : :triste: ;

  4. #4
    Expert éminent
    Homme Profil pro
    Architecte technique retraité
    Inscrit en
    Juin 2008
    Messages
    21 683
    Détails du profil
    Informations personnelles :
    Sexe : Homme
    Localisation : France, Manche (Basse Normandie)

    Informations professionnelles :
    Activité : Architecte technique retraité
    Secteur : Industrie

    Informations forums :
    Inscription : Juin 2008
    Messages : 21 683
    Par défaut
    Citation Envoyé par papajoker Voir le message
    PS : Pour l'application graphique (TK-APP), j'ai demandé à une IA de me l'écrire, comme le protocole socket.
    Ca se voit!
    Utilisez plutôt le module logging qui est thread safe et spécialiser un widget Text pour être logging.Handler.
    Une daube "majoritaire" reste une daube et votre truc ne fait pas rêver (car ils y a un tas de crétins qui vont croire qu'ils peuvent déconnecter leur cerveau - et vous n'y êtes pour rien) .

    - W
    Architectures post-modernes.
    Python sur DVP c'est aussi des FAQs, des cours et tutoriels

  5. #5
    Expert confirmé Avatar de papajoker
    Homme Profil pro
    Développeur Web
    Inscrit en
    Septembre 2013
    Messages
    2 291
    Détails du profil
    Informations personnelles :
    Sexe : Homme
    Localisation : France, Nièvre (Bourgogne)

    Informations professionnelles :
    Activité : Développeur Web
    Secteur : High Tech - Multimédia et Internet

    Informations forums :
    Inscription : Septembre 2013
    Messages : 2 291
    Par défaut
    vous n'y êtes pour rien
    si si, c'est bien moi le fautif ! C'est quand même ma requête à l'ia !

    Je ne connais que légèrement tk. Je désirais une app sans librairies "externes" (pour gui et communication) et j'ai demandé un code minimal. J'estime que je n'ai un "juste" retour des choses. Comme toujours avec l'ia, c'est à moi d'être plus intelligent dans ma demande.

    Après, il faut bien voir que dans ce message, je suis dans la rêverie. Piloter une app est trop complexe ! Ok pour les actions du menu principal (ce qui est déjà bien).
    Mais ensuite, nous n'avons pas un problème MCP mais classique pour toute app gui. Si je suis dans un de mes 36 dialogues, il me faut une commande (externe) pour sélectionner le "nom", pour modifier l'age, ... c'est à l'infini. Combien de commande pour "sélectionner xxx dans le combo Y dans le dialogue Z" ? Et comment distinguer des commandes "calculer".. "mettre à jour" qui sont diverses commandes en fonction du contexte de l'app (dialogue modif client ou ... dialogue add produit ?)

    Il n'était donc, pour moi, pas impératif d'avoir un code proche de la production, juste une preuve de concept. Si un client demande, j'ai déjà une idée ou je mets les pieds.
    $moi= (:nono: !== :oops:) ? :king: : :triste: ;

Discussions similaires

  1. Nouvelle version de MySql
    Par syannic dans le forum SQL Procédural
    Réponses: 8
    Dernier message: 17/03/2003, 17h39
  2. [VB6] Ouverture d'une nouvelle fenêtre dans un MDI
    Par pepper dans le forum VB 6 et antérieur
    Réponses: 8
    Dernier message: 17/02/2003, 14h03
  3. [XMLRAD] Nouvelle V7
    Par rgarnier dans le forum XMLRAD
    Réponses: 3
    Dernier message: 15/01/2003, 09h57

Partager

Partager
  • Envoyer la discussion sur Viadeo
  • Envoyer la discussion sur Twitter
  • Envoyer la discussion sur Google
  • Envoyer la discussion sur Facebook
  • Envoyer la discussion sur Digg
  • Envoyer la discussion sur Delicious
  • Envoyer la discussion sur MySpace
  • Envoyer la discussion sur Yahoo