test

app.py

@@ -1,68 +1,83 @@
-import gradio as gr
-from openai import OpenAI
+"""
+cd to the `examples/snippets/clients` directory and run:
+    uv run client
+"""
+
+import asyncio
 import os
-import json
-import requests
-
-cle_api = os.environ.get("CLE_API_MISTRAL")
-MCP_URL = "https://hackathoncra-gradio-mcp.hf.space/gradio_api/mcp/"
-
-# Initialisation du client Mistral (API compatible OpenAI)
-clientLLM = OpenAI(api_key=cle_api, base_url="https://api.mistral.ai/v1")
-
-def call_mcp(payload: dict):
-    """Appel simple au serveur MCP Gradio"""
-    headers = {"Content-Type": "application/json"}
-    response = requests.post(MCP_URL, data=json.dumps(payload), headers=headers)
-    print(response.json())
-    response.raise_for_status()
-    return response.json()
-
-# Chatbot avec Mistral + MCP
-def chatbot(message, history):
-    # Préparer l’historique pour Mistral
-    messages = []
-    for user_msg, bot_msg in history:
-        messages.append({"role": "user", "content": user_msg})
-        messages.append({"role": "assistant", "content": bot_msg})
-    
-    messages.append({"role": "user", "content": message})
-
-    # Appel API Mistral
-    response = clientLLM.chat.completions.create(
-        model="mistral-small-latest",
-        messages=messages
+
+from pydantic import AnyUrl
+
+from mcp import ClientSession, StdioServerParameters, types
+from mcp.client.stdio import stdio_client
+from mcp.shared.context import RequestContext
+
+# Create server parameters for stdio connection
+server_params = StdioServerParameters(
+    command="uv",  # Using uv to run the server
+    args=["run", "server", "fastmcp_quickstart", "stdio"],  # We're already in snippets dir
+    env={"UV_INDEX": os.environ.get("UV_INDEX", "")},
+)
+
+
+# Optional: create a sampling callback
+async def handle_sampling_message(
+    context: RequestContext[ClientSession, None], params: types.CreateMessageRequestParams
+) -> types.CreateMessageResult:
+    print(f"Sampling request: {params.messages}")
+    return types.CreateMessageResult(
+        role="assistant",
+        content=types.TextContent(
+            type="text",
+            text="Hello, world! from model",
+        ),
+        model="gpt-3.5-turbo",
+        stopReason="endTurn",
     )
 
-    bot_reply = response.choices[0].message.content.strip()
 
-    # Vérifier si la réponse contient un JSON MCP
-    try:
-        mcp_payload = json.loads(bot_reply)
-        mcp_result = call_mcp(mcp_payload)
-        bot_reply = f"Réponse via MCP:\n{json.dumps(mcp_result, indent=2)}"
-    except json.JSONDecodeError:
-        # Pas de JSON MCP, réponse normale
-        pass
+async def run():
+    async with stdio_client(server_params) as (read, write):
+        async with ClientSession(read, write, sampling_callback=handle_sampling_message) as session:
+            # Initialize the connection
+            await session.initialize()
+
+            # List available prompts
+            prompts = await session.list_prompts()
+            print(f"Available prompts: {[p.name for p in prompts.prompts]}")
+
+            # Get a prompt (greet_user prompt from fastmcp_quickstart)
+            if prompts.prompts:
+                prompt = await session.get_prompt("greet_user", arguments={"name": "Alice", "style": "friendly"})
+                print(f"Prompt result: {prompt.messages[0].content}")
+
+            # List available resources
+            resources = await session.list_resources()
+            print(f"Available resources: {[r.uri for r in resources.resources]}")
+
+            # List available tools
+            tools = await session.list_tools()
+            print(f"Available tools: {[t.name for t in tools.tools]}")
 
-    history.append(("Vous: " + message, "Bot: " + bot_reply))
-    return history, history
+            # Read a resource (greeting resource from fastmcp_quickstart)
+            resource_content = await session.read_resource(AnyUrl("greeting://World"))
+            content_block = resource_content.contents[0]
+            if isinstance(content_block, types.TextContent):
+                print(f"Resource content: {content_block.text}")
 
+            # Call a tool (add tool from fastmcp_quickstart)
+            result = await session.call_tool("add", arguments={"a": 5, "b": 3})
+            result_unstructured = result.content[0]
+            if isinstance(result_unstructured, types.TextContent):
+                print(f"Tool result: {result_unstructured.text}")
+            result_structured = result.structuredContent
+            print(f"Structured tool result: {result_structured}")
 
-def call_mcp(payload: dict):
-    """
-    Fonction générique pour interroger le serveur MCP hébergé sur Gradio.
-    """
-    headers = {"Content-Type": "application/json"}
-    response = requests.post(MCP_URL, data=json.dumps(payload), headers=headers)
-    response.raise_for_status()
-    return response.json()
-    
-with gr.Blocks() as demo:
 
-    chatbot_ui = gr.Chatbot(label="ChatBot")
-    msg = gr.Textbox(placeholder="Écrivez un message...")
+def main():
+    """Entry point for the client script."""
+    asyncio.run(run())
 
-    msg.submit(chatbot, [msg, chatbot_ui], [chatbot_ui, chatbot_ui])
 
-demo.launch(debug=True)
+if __name__ == "__main__":
+    main()