uv sync and rep[lacing openai with requests

Author: yarnabrina

examples/clients/simple-sampling-client/.env.example

@@ -1 +1 @@
-OPENAI_API_KEY=YOUR_OPENAI_API_KEY
+API_KEY=YOUR_API_KEY

examples/clients/simple-sampling-client/mcp_simple_sampling_client/main.py

@@ -3,17 +3,10 @@
 import asyncio
 import http
 import json
+import typing
 
-import openai
 import pydantic_settings
-from openai.types.chat import (
-    ChatCompletion,
-    ChatCompletionAssistantMessageParam,
-    ChatCompletionMessageParam,
-    ChatCompletionSystemMessageParam,
-    ChatCompletionToolParam,
-    ChatCompletionUserMessageParam,
-)
+import requests
 
 import mcp
 from mcp.client.streamable_http import streamablehttp_client
@@ -24,11 +17,12 @@
 class Configurations(pydantic_settings.BaseSettings):
     """Define configurations for the sampling client."""
 
+    base_url: str = "https://api.openai.com/v1"
+    api_key: str = "your_api_key"
     chat_model: str = "gpt-4o-mini"
     max_tokens: int = 1024
     mcp_server_host: str = "localhost"
     mcp_server_port: int = 8000
-    openai_api_key: str = "your_openai_api_key"
     system_prompt: str = "You are a helpful assistant."
 
     model_config = pydantic_settings.SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
@@ -47,50 +41,58 @@ def __init__(self: "SamplingClient", config: Configurations) -> None:
         self.config = config
 
         self.server_url = f"http://{self.config.mcp_server_host}:{self.config.mcp_server_port}/mcp"
-        self.openai_client = openai.OpenAI(api_key=self.config.openai_api_key)
+        self.api_url = f"{self.config.base_url}/chat/completions"
 
-        self.conversation_history: list[ChatCompletionMessageParam] = []
+        self.conversation_history: list[dict[str, str]] = []
 
-    def get_openai_response(
+    def get_llm_response(
         self: "SamplingClient",
-        chat_history: list[ChatCompletionMessageParam],
+        chat_history: list[dict[str, str]],
         system_prompt: str,
         max_tokens: int,
-        tools: list[ChatCompletionToolParam] | None = None,
-    ) -> ChatCompletion:
-        """Get a non-streaming response from OpenAI API.
+        tools: list[dict[str, typing.Any]] | None = None,
+    ) -> dict[str, typing.Any]:
+        """Get a non-streaming response from OpenAI compatible LLM API.
 
         Parameters
         ----------
-        chat_history : list[ChatCompletionMessageParam]
+        chat_history : list[dict[str, str]]
             The chat history to use for the chat completion.
         system_prompt : str
             The system prompt to use for the chat completion.
         max_tokens : int
             The maximum number of tokens to generate in the response.
-        tools : list[ChatCompletionToolParam] | None, optional
+        tools : list[dict[str, typing.Any]] | None, optional
             The tools to use for the chat completion, by default None.
 
         Returns
         -------
-        ChatCompletion
-            The response from the OpenAI API.
+        dict[str, typing.Any]
+            The response from the LLM API.
         """
         updated_chat_history = [
-            ChatCompletionSystemMessageParam(content=system_prompt, role="system"),
+            {"content": system_prompt, "role": "system"},
             *chat_history,
         ]
 
         extra_arguments = {} if tools is None else {"tool_choice": "auto", "tools": tools}
 
-        chat_completion = self.openai_client.chat.completions.create(
-            messages=updated_chat_history,
-            model=self.config.chat_model,
-            max_completion_tokens=max_tokens,
-            n=1,
-            stream=False,
-            **extra_arguments,
-        )
+        chat_completion = requests.post(
+            self.api_url,
+            json={
+                "messages": updated_chat_history,
+                "model": self.config.chat_model,
+                "max_completion_tokens": max_tokens,
+                "n": 1,
+                "stream": False,
+                **extra_arguments,
+            },
+            headers={
+                "Authorization": f"Bearer {self.config.api_key}",
+                "Content-Type": "application/json",
+            },
+            timeout=300,
+        ).json()
 
         return chat_completion
 
@@ -105,7 +107,7 @@ async def fetch_mcp_tools(self: "SamplingClient") -> list[Tool]:
         return server_tools.tools
 
     @staticmethod
-    def convert_to_openai_tools(mcp_tools: list[Tool]) -> list[ChatCompletionToolParam]:
+    def convert_to_openai_tools(mcp_tools: list[Tool]) -> list[dict[str, typing.Any]]:
         """Convert MCP tools to OpenAI tool call parameters.
 
         Parameters
@@ -115,29 +117,25 @@ def convert_to_openai_tools(mcp_tools: list[Tool]) -> list[ChatCompletionToolPar
 
         Returns
         -------
-        list[ChatCompletionToolParam]
+        list[dict[str, typing.Any]]
             List of OpenAI tool call parameters.
         """
         return [
-            ChatCompletionToolParam(
-                function={
-                    "name": tool.name,
-                    "description": tool.description or "",
-                    "parameters": tool.inputSchema,
-                },
-                type="function",
-            )
+            {
+                "function": {"name": tool.name, "description": tool.description or "", "parameters": tool.inputSchema},
+                "type": "function",
+            }
             for tool in mcp_tools
         ]
 
     async def sampling_handler(
-        self: "SamplingClient", context: RequestContext, parameters: CreateMessageRequestParams
+        self: "SamplingClient", context: RequestContext[typing.Any, typing.Any], parameters: CreateMessageRequestParams
     ) -> CreateMessageResult | ErrorData:
         """Handle sampling requests for OpenAI API calls with MCP tools.
 
         Parameters
         ----------
-        context : RequestContext
+        context : RequestContext[typing.Any, typing.Any]
             request context containing information about the sampling request
         parameters : CreateMessageRequestParams
             parameters for the sampling request, including messages and customisations
@@ -149,44 +147,44 @@ async def sampling_handler(
         """
         del context
 
-        openai_response = self.get_openai_response(
+        openai_response = self.get_llm_response(
             [
-                ChatCompletionUserMessageParam(
-                    content=(
-                        message.content.text if isinstance(message.content, TextContent) else str(message.content)
-                    ),
-                    role="user",
-                )
+                {
+                    "content": message.content.text
+                    if isinstance(message.content, TextContent)
+                    else str(message.content),
+                    "role": "user",
+                }
                 for message in parameters.messages
             ],
             parameters.systemPrompt or self.config.system_prompt,
             parameters.maxTokens,
         )
 
-        if not (choices := openai_response.choices):
+        if not (choices := openai_response["choices"]):
             return ErrorData(
                 code=http.HTTPStatus.INTERNAL_SERVER_ERROR,
                 message="No choices returned from OpenAI API.",
             )
 
         choice = choices[0]
-        sampling_response_message = choice.message.content or ""
+        sampling_response_message = choice["message"]["content"] or ""
 
         return CreateMessageResult(
             role="assistant",
             content=TextContent(type="text", text=sampling_response_message),
             model=self.config.chat_model,
-            stopReason=choice.finish_reason,
+            stopReason=choice["finish_reason"],
         )
 
-    async def execute_tool_call(self: "SamplingClient", tool_name: str, arguments: dict) -> str:
+    async def execute_tool_call(self: "SamplingClient", tool_name: str, arguments: dict[str, typing.Any]) -> str:
         """Execute a tool call on an MCP server.
 
         Parameters
         ----------
         tool_name : str
             name of the tool to call, formatted as "mcp-{server_name}-{tool_name}"
-        arguments : dict
+        arguments : dict[str, typing.Any]
             arguments to pass to the tool call
 
         Returns
@@ -212,69 +210,62 @@ async def execute_tool_call(self: "SamplingClient", tool_name: str, arguments: d
 
     async def orchestrate(self: "SamplingClient", user_message: str) -> None:
         """Orchestrate the sampling client to handle requests."""
-        self.conversation_history.append(ChatCompletionUserMessageParam(role="user", content=user_message))
+        self.conversation_history.append({"role": "user", "content": user_message})
 
         self.mcp_server_tools = await self.fetch_mcp_tools()
         self.openai_compatible_tools = self.convert_to_openai_tools(self.mcp_server_tools)
 
-        openai_response = self.get_openai_response(
+        openai_response = self.get_llm_response(
             self.conversation_history,
             self.config.system_prompt,
             self.config.max_tokens,
             tools=self.openai_compatible_tools,
         )
 
-        if not (choices := openai_response.choices):
+        if not (choices := openai_response["choices"]):
             error_message = "No choices returned from OpenAI API."
-            self.conversation_history.append(
-                ChatCompletionAssistantMessageParam(role="assistant", content=error_message)
-            )
+            self.conversation_history.append({"role": "assistant", "content": error_message})
 
             print(error_message)
 
             return
 
         choice = choices[0]
 
-        while choice.finish_reason == "tool_calls":
-            for tool_call in choice.message.tool_calls or []:
-                if tool_call.type != "function":
+        while choice["finish_reason"] == "tool_calls":
+            for tool_call in choice["message"]["tool_calls"] or []:
+                if tool_call["type"] != "function":
                     continue
 
                 tool_response = await self.execute_tool_call(
-                    tool_call.function.name, json.loads(tool_call.function.arguments)
+                    tool_call["function"]["name"], json.loads(tool_call["function"]["arguments"])
                 )
 
                 self.conversation_history.append(
-                    ChatCompletionAssistantMessageParam(
-                        role="assistant",
-                        content=f"Tool {tool_call.id} returned: {tool_response}",
-                    )
+                    {"role": "assistant", "content": f"Tool {tool_call['id']} returned: {tool_response}"}
                 )
 
-            openai_response = self.get_openai_response(
+            openai_response = self.get_llm_response(
                 self.conversation_history,
                 self.config.system_prompt,
                 self.config.max_tokens,
                 tools=self.openai_compatible_tools,
             )
 
-            if not (choices := openai_response.choices):
+            if not (choices := openai_response["choices"]):
                 error_message = "No choices returned from OpenAI API."
-                self.conversation_history.append(
-                    ChatCompletionAssistantMessageParam(role="assistant", content=error_message)
-                )
+                self.conversation_history.append({"role": "assistant", "content": error_message})
 
                 print(error_message)
 
                 return
 
             choice = choices[0]
 
-        assistant_message = choice.message.content or ""
+        assistant_message = choice["message"]["content"] or ""
         self.conversation_history.append({"role": "assistant", "content": assistant_message})
 
-        print(f"Assistant: {choice.message.content}")
+        print(f"Assistant: {assistant_message}")
 
 
 def main():

examples/clients/simple-sampling-client/pyproject.toml

@@ -38,9 +38,9 @@ classifiers = [
 ]
 dependencies = [
   "mcp>=1.16.0,<2",
-  "openai>=2.1.0,<3",
   "pydantic>=2.11.10,<3",
   "pydantic-settings>=2.11.0,<3",
+  "requests>=2.32.5,<3",
 ]
 
 [project.scripts]