feat: update extra code

Author: aac228

examples/async_chat_with_image_no_streaming.py

@@ -10,7 +10,7 @@
 
 async def main():
     api_key = os.environ["MISTRAL_API_KEY"]
-    model = "pixtral-12b"
+    model = "pixtral-12b-2409"
     client = Mistral(api_key=api_key)
 
     chat_response = await client.chat.complete_async(
@@ -21,7 +21,7 @@ async def main():
                     {"type": "text", "text": "What's in this image?"},
                     {
                         "type": "image_url",
-                        "image_url": "https://cms.mistral.ai/assets/af26a11d-0793-439f-a06e-7694b24b8270",
+                        "image_url": "https://cms.mistral.ai/assets/a64b3821-3a4c-4d4d-b718-d653f3eb7a5e.png?",
                     },
                 ]
             )

examples/async_chat_with_streaming.py

@@ -20,6 +20,7 @@ async def main():
             UserMessage(content="What is the best French cheese?give the best 50")
         ],
     )
+    assert response
     async for chunk in response:
         if chunk.data.choices[0].delta.content is not None:
             print(chunk.data.choices[0].delta.content, end="")

examples/async_conversation_run.py

@@ -0,0 +1,57 @@
+#!/usr/bin/env python
+import asyncio
+import os
+
+from mistralai import Mistral
+from mistralai.extra.run.context import RunContext
+from mistralai.types import BaseModel
+
+MODEL = "mistral-medium-latest"
+
+
+def math_question_generator(question_num: int):
+    """Random generator of mathematical question
+
+    Args:
+        question_num (int): the number of the question that will be returned, should be between 1-100
+    """
+    return (
+        "solve the following differential equation: `y'' + 3y' + 2y = 0`"
+        if question_num % 2 == 0
+        else "solve the following differential equation: `y'' - 4y' + 4y = e^x`"
+    )
+
+
+async def main():
+    api_key = os.environ["MISTRAL_API_KEY"]
+    client = Mistral(api_key=api_key)
+
+    class Explanation(BaseModel):
+        explanation: str
+        output: str
+
+    class MathDemonstration(BaseModel):
+        steps: list[Explanation]
+        final_answer: str
+
+    async with RunContext(model=MODEL, output_format=MathDemonstration) as run_ctx:
+        # register a new function that can be executed on the client side
+        run_ctx.register_func(math_question_generator)
+        run_result = await client.beta.conversations.run_async(
+            run_ctx=run_ctx,
+            instructions="Use the code interpreter to help you when asked mathematical questions.",
+            inputs=[
+                {"role": "user", "content": "hey"},
+                {"role": "assistant", "content": "hello"},
+                {"role": "user", "content": "Request a math question and answer it."},
+            ],
+            tools=[{"type": "code_interpreter"}],
+        )
+        print("All run entries:")
+        for entry in run_result.output_entries:
+            print(f"{entry}")
+        print(f"Final model: {run_result.output_as_model}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/async_conversation_run_mcp.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python
+import asyncio
+import os
+import random
+
+from mistralai import Mistral
+from mistralai.extra.run.context import RunContext
+from mcp import StdioServerParameters
+from mistralai.extra.mcp.stdio import (
+    MCPClientSTDIO,
+)
+from pathlib import Path
+
+from mistralai.types import BaseModel
+
+cwd = Path(__file__).parent
+MODEL = "mistral-medium-latest"
+
+
+async def main() -> None:
+    api_key = os.environ["MISTRAL_API_KEY"]
+    client = Mistral(api_key=api_key)
+
+    # Create a mcp server has a tool to return the weather based on the location
+    server_params = StdioServerParameters(
+        command="python",
+        args=[str((cwd / "mcp_servers/stdio_server.py").resolve())],
+        env=None,
+    )
+
+    weather_agent = client.beta.agents.create(
+        model=MODEL,
+        name="weather teller",
+        instructions="You are able to tell the weather.",
+        description="",
+    )
+
+    class WeatherResult(BaseModel):
+        user: str
+        location: str
+        temperature: float
+
+    async with RunContext(
+        agent_id=weather_agent.id,
+        output_format=WeatherResult,
+        continue_on_fn_error=True,
+    ) as run_ctx:
+        # Add location function to the run context
+        @run_ctx.register_func
+        def get_location(name: str) -> str:
+            """function to get location of a user.
+
+            Args:
+                name: name of the user.
+            """
+            return random.choice(["New York", "London", "Paris", "Tokyo", "Sydney"])
+
+        # Add mcp client to the run context
+        mcp_client = MCPClientSTDIO(stdio_params=server_params)
+        await run_ctx.register_mcp_client(mcp_client=mcp_client)
+
+        run_result = await client.beta.conversations.run_async(
+            run_ctx=run_ctx,
+            inputs="Tell me the weather in John's location currently.",
+        )
+
+        print("All run entries:")
+        for entry in run_result.output_entries:
+            print(f"{entry}")
+            print()
+        print(f"Final model: {run_result.output_as_model}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/async_conversation_run_mcp_remote.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python
+import asyncio
+import os
+
+from mistralai import Mistral
+from mistralai.extra.run.context import RunContext
+
+from mistralai.extra.mcp.sse import (
+    MCPClientSSE,
+    SSEServerParams,
+)
+from pathlib import Path
+
+cwd = Path(__file__).parent
+MODEL = "mistral-medium-latest"
+
+# Use an official remote mcp server
+# you can find some at:
+# - https://mcpservers.org/remote-mcp-servers
+# this one does not require auth: https://remote.mcpservers.org/edgeone-pages/mcp
+
+
+async def main():
+    api_key = os.environ["MISTRAL_API_KEY"]
+    client = Mistral(api_key=api_key)
+
+    server_url = "https://mcp.semgrep.ai/sse"
+    mcp_client = MCPClientSSE(sse_params=SSEServerParams(url=server_url, timeout=100))
+
+    async with RunContext(
+        model=MODEL,
+    ) as run_ctx:
+        # Add mcp client to the run context
+        await run_ctx.register_mcp_client(mcp_client=mcp_client)
+
+        run_result = await client.beta.conversations.run_async(
+            run_ctx=run_ctx,
+            inputs="Can you write a hello_world.py and check for security vulnerabilities",
+        )
+
+        print("All run entries:")
+        for entry in run_result.output_entries:
+            print(f"{entry}")
+            print()
+        print(f"Final Response: {run_result.output_as_text}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/async_conversation_run_mcp_remote_auth.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python
+import asyncio
+from http.server import BaseHTTPRequestHandler, HTTPServer
+import os
+import threading
+import webbrowser
+
+from mistralai import Mistral
+from mistralai.extra.run.context import RunContext
+
+from mistralai.extra.mcp.sse import (
+    MCPClientSSE,
+    SSEServerParams,
+)
+from mistralai.extra.mcp.auth import build_oauth_params
+
+MODEL = "mistral-medium-latest"
+
+CALLBACK_PORT = 16010
+
+
+# Use an official remote mcp server
+# you can find some at:
+# - https://mcpservers.org/remote-mcp-servers
+# - https://support.anthropic.com/en/articles/11176164-pre-built-integrations-using-remote-mcp
+# this one has auth: https://mcp.linear.app/sse
+
+
+def run_callback_server(callback_func):
+    auth_response: dict = {"url": ""}
+
+    class OAuthCallbackHandler(BaseHTTPRequestHandler):
+        server_version = "HTTP"
+        code = None
+
+        def do_GET(self):
+            if "/callback" in self.path:
+                try:
+                    auth_response["url"] = self.path
+                    self.send_response(200)
+                    self.send_header("Content-type", "text/html")
+                    self.end_headers()
+                    callback_func()
+                    response_html = "<html><body><p>You may now close this window.</p></body></html>"
+                    self.wfile.write(response_html.encode())
+                    threading.Thread(target=httpd.shutdown).start()
+                except Exception:
+                    self.send_response(500)
+                    self.end_headers()
+
+    server_address = ("localhost", CALLBACK_PORT)
+    httpd = HTTPServer(server_address, OAuthCallbackHandler)
+    threading.Thread(target=httpd.serve_forever).start()
+    redirect_url = f"http://localhost:{CALLBACK_PORT}/oauth/callback"
+    return httpd, redirect_url, auth_response
+
+
+async def main():
+    api_key = os.environ["MISTRAL_API_KEY"]
+    client = Mistral(api_key=api_key)
+
+    server_url = "https://mcp.linear.app/sse"
+
+    # set-up the client
+    mcp_client = MCPClientSSE(
+        sse_params=SSEServerParams(
+            url=server_url,
+        )
+    )
+
+    callback_event = asyncio.Event()
+    event_loop = asyncio.get_event_loop()
+
+    # check if auth is required
+    if await mcp_client.requires_auth():
+        # let's login
+        httpd, redirect_url, auth_response = run_callback_server(
+            callback_func=lambda: event_loop.call_soon_threadsafe(callback_event.set)
+        )
+        try:
+            # First create the required oauth config, this means fetching the server metadata and registering a client
+            oauth_params = await build_oauth_params(
+                mcp_client.base_url, redirect_url=redirect_url
+            )
+            mcp_client.set_oauth_params(oauth_params=oauth_params)
+            login_url, state = await mcp_client.get_auth_url_and_state(redirect_url)
+
+            # The oauth params like client_id, client_secret would generally be saved in some persistent storage.
+            # The oauth state and token would be saved in a user session.
+
+            # wait for the user to complete the authentication process
+            print("Please go to this URL and authorize the application:", login_url)
+            webbrowser.open(login_url, new=2)
+            await callback_event.wait()
+
+            # in a real app this would be your oauth2 callback route you would get the code from the query params,
+            # verify the state, and then get the token
+            # Here we recreate a new client with the saved params which and exchange the code for a token
+            mcp_client = MCPClientSSE(
+                sse_params=SSEServerParams(
+                    url=server_url,
+                ),
+                oauth_params=oauth_params,
+            )
+
+            token = await mcp_client.get_token_from_auth_response(
+                auth_response["url"], redirect_url=redirect_url, state=state
+            )
+            mcp_client.set_auth_token(token)
+
+        except Exception as e:
+            print(f"Error during authentication: {e}")
+        finally:
+            httpd.shutdown()
+            httpd.server_close()
+
+    # Now it's possible to make a query to the mcp server as we would do without authentication
+    async with RunContext(
+        model=MODEL,
+    ) as run_ctx:
+        # Add mcp client to the run context
+        await run_ctx.register_mcp_client(mcp_client=mcp_client)
+
+        run_result = await client.beta.conversations.run_async(
+            run_ctx=run_ctx,
+            inputs="Tell me which projects do I have in my workspace?",
+        )
+
+        print(f"Final Response: {run_result.output_as_text}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())