OpenAI MCP Server

Instructions

Implementation Reference

• src/mcp_server_openai/server.py:21-39 (registration)

  Registers the "ask-openai" tool with MCP server, including its description and input schema definition.

  @server.list_tools()
  async def handle_list_tools() -> list[types.Tool]:
      return [
          types.Tool(
              name="ask-openai",
              description="Ask my assistant models a direct question",
              inputSchema={
                  "type": "object",
                  "properties": {
                      "query": {"type": "string", "description": "Ask assistant"},
                      "model": {"type": "string", "default": "gpt-4", "enum": ["gpt-4", "gpt-3.5-turbo"]},
                      "temperature": {"type": "number", "default": 0.7, "minimum": 0, "maximum": 2},
                      "max_tokens": {"type": "integer", "default": 500, "minimum": 1, "maximum": 4000}
                  },
                  "required": ["query"]
              }
          )
      ]

• src/mcp_server_openai/server.py:40-59 (handler)

  Handles tool calls, specifically implementing the logic for "ask-openai" by parsing arguments and delegating to LLMConnector.ask_openai.

  @server.call_tool()
  async def handle_tool_call(name: str, arguments: dict | None) -> list[types.TextContent]:
      try:
          if not arguments:
              raise ValueError("No arguments provided")
  
          if name == "ask-openai":
              response = await connector.ask_openai(
                  query=arguments["query"],
                  model=arguments.get("model", "gpt-4"),
                  temperature=arguments.get("temperature", 0.7),
                  max_tokens=arguments.get("max_tokens", 500)
              )
              return [types.TextContent(type="text", text=f"OpenAI Response:\n{response}")]
  
          raise ValueError(f"Unknown tool: {name}")
      except Exception as e:
          logger.error(f"Tool call failed: {str(e)}")
          return [types.TextContent(type="text", text=f"Error: {str(e)}")]

• src/mcp_server_openai/llm.py:10-24 (helper)

  Core implementation of the OpenAI query logic used by the "ask-openai" tool, performing the actual API call.

  async def ask_openai(self, query: str, model: str = "gpt-4", temperature: float = 0.7, max_tokens: int = 500) -> str:
      try:
          response = await self.client.chat.completions.create(
              messages=[
                  {"role": "system", "content": "You are a helpful assistant."},
                  {"role": "user", "content": query}
              ],
              model=model,
              temperature=temperature,
              max_tokens=max_tokens
          )
          return response.choices[0].message.content
      except Exception as e:
          logger.error(f"Failed to query OpenAI: {str(e)}")
          raise