@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"

#!/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

#!/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))

(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

(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

#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)