Build your own MCP Server and MCP Client from Scratch

class MCPClient:
  def __init__(self):
      self.session: Optional[ClientSession] = None
      self.exit_stack = AsyncExitStack()
      self.openai = OpenAI()

  async def connect_to_server(self, server_script_path: str):
      """Connect to an MCP server"""
      server_params = StdioServerParameters(
          command="python",
          args=[server_script_path],
          env=None
      )
     
      stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
      self.stdio, self.write = stdio_transport
      self.session = await

self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
     
      await self.session.initialize()
     
      response = await self.session.list_tools()
      tools = response.tools
      print("\nConnected to server with tools:", [tool.name for tool in tools])

  async def process_query(self, query: str) -> str:
      """Process a query using OpenAI and available tools"""
      messages = [{
              "role": "user",
              "content": query
          }]

      response = await self.session.list_tools()
      available_tools = [{
          "type": "function",
          "function": {
              "name": tool.name,
              "description": tool.description,
              "parameters": tool.inputSchema
          }
      } for tool in response.tools]

      response = self.openai.chat.completions.create(
          model="gpt-4o-mini",
          messages=messages,
          tools=available_tools
      )

      final_text = []
      message = response.choices[0].message

      if message.content:
          final_text.append(message.content)

      if message.tool_calls:
          messages.append({
              "role": "assistant",
              "content": message.content,
              "tool_calls": message.tool_calls

           })

          for tool_call in message.tool_calls:
              tool_name = tool_call.function.name
              tool_args = eval(tool_call.function.arguments)  # Parse JSON string to dict
             
              result = await self.session.call_tool(tool_name, tool_args)
              final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")

              messages.append({
                  "role": "tool",
                  "tool_call_id": tool_call.id,
                  "content": str(result.content)
              })

          response = self.openai.chat.completions.create(
              model="gpt-4o-mini",
              messages=messages,
          )

          final_text.append(response.choices[0].message.content)

      return "\n".join(final_text)

  async def chat_loop(self):
      """Run an interactive chat loop"""
      print("\nMCP Client Started!")
      print("Type your queries or 'quit' to exit.")
     
      while True:
          try:
              query = input("\nQuery: ").strip()
              if query.lower() == 'quit':
                  break
                 
              response = await self.process_query(query)
              print("\n" + response)
                 
          except Exception as e:
              print(f"\nError: {str(e)}")
 
  async def cleanup(self):
      """Clean up resources"""
      await self.exit_stack.aclose()