If I want to use a python client to call this MCP service, what format do I need to pass to this MCP service?

[MrKyle0315]

import asyncio
import os
import re
from openai import OpenAI
from dotenv import load_dotenv
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

Load environment variables

load_dotenv()

class MCPClient:
def init(self):
"""Initialize the MCP client."""
self.exit_stack = AsyncExitStack()
self.openai_api_key = os.getenv("OPENAI_API_KEY")
self.base_url = os.getenv("BASE_URL")
self.model = os.getenv("MODEL")

    if not self.openai_api_key:
        raise ValueError("OpenAI API Key not found. Please set OPENAI_API_KEY in the .env file.")
    
    self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)
    self.session = None
    # Stores the comma-separated list of available tool names from the MCP service
    self.available_tool_names = ""

async def connect_to_mcp_server(self):
    """Connect to the MCP server and list available tools."""
    try:
        server_params = StdioServerParameters(
            command="npx",
            args=[ 
                "--directory",
                "E:/deeplearn/mcp-playwright",  # Change to the correct directory path
                "run",
                "@modelcontextprotocol/server-playwright"
            ]
        )
        
        # Connect to the server using stdio_client
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                
                # Get available tools
                response = await session.list_tools()
                tools = response.tools
                
                print("\nConnected to the server. The following tools are available:")
                tool_names = []
                for tool in tools:
                    tool_names.append(tool.name)
                    print(f"- {tool.name}: {tool.description}")
                
                # Dynamically build the list of tool names for limiting model calls
                self.available_tool_names = ", ".join(tool_names)
                
                # Save the session for later use
                self.stdio = read
                self.write = write
                self.session = session
                
                return tools
    except Exception as e:
        print(f"An error occurred while connecting to the MCP server: {e}")
        raise

async def process_query(self, query: str) -> str:
    """
    Call the OpenAI API to process a user query and generate a series of tool call commands
    based on the model's response, ensuring that the model selects only from the currently available tools.
    """
    # Construct the system prompt with the available tool names dynamically
    system_prompt = (
        "Please generate a series of tool call commands based on the user's description. "
        "Each command should be output in the format call_tool: <tool_name>(param=value) and do not return any extra descriptive text. "
        "Ensure that <tool_name> is chosen only from the following available tools: "
        f"{self.available_tool_names}"
    )
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": query}
    ]
    try:
        # Call the OpenAI API to get a response
        response = await asyncio.get_event_loop().run_in_executor(
            None,
            lambda: self.client.chat.completions.create(
                model=self.model,
                messages=messages
            )
        )
        ai_response = response.choices[0].message.content.strip()
        print(f"AI response (commands or answer): {ai_response}")
    except Exception as e:
        return f"⚠️ An error occurred when calling the OpenAI API: {str(e)}"

    # Split the response into lines and process each tool call command
    lines = ai_response.splitlines()
    results = []
    non_tool_lines = []
    for line in lines:
        line = line.strip()
        if line.lower().startswith("call_tool:"):
            try:
                # Use regex to parse the tool name and parameter string
                match = re.match(r"call_tool:\s*(\w+)\((.*?)\)", line, re.IGNORECASE)
                if not match:
                    results.append("⚠️ Unable to parse the tool call command. Please check the format.")
                    continue
                tool_name = match.group(1).strip()
                params_str = match.group(2).strip()
                params = {}
                if params_str:
                    # Use regex to parse key-value pairs, allowing commas in the parameter values
                    # Matches patterns like key="value", or key='value', or key=value (without comma)
                    pattern = re.compile(r'(\w+)\s*=\s*(?:"([^"]*)"|\'([^\']*)\'|([^,]+))')
                    for m in pattern.findall(params_str):
                        key = m[0]
                        value = m[1] or m[2] or m[3]
                        params[key.strip()] = value.strip()
                print(f"Parsed tool call: tool name = {tool_name}, parameters = {params}")
                # Call the tool from the MCP service
                tool_result = await self.session.call_tool(tool_name, params)
                # Assuming tool_result has a content attribute
                results.append(f"Result of executing tool '{tool_name}': {tool_result.content}")
            except Exception as e:
                results.append(f"⚠️ Tool call failed: {str(e)}")
        else:
            non_tool_lines.append(line)
    if results:
        return "\n".join(results)
    else:
        return "\n".join(non_tool_lines)

async def chat_loop(self):
    """Run an interactive chat loop."""
    # Connect to the MCP server and get the list of available tools first
    await self.connect_to_mcp_server()
    print("\n🤖 MCP client started! Type 'quit' to exit.")
    while True:
        try:
            user_input = input("\nYou: ").strip()
            if user_input.lower() in ['quit', 'exit', 'q']:
                break

            response = await self.process_query(user_input)
            print(f"\n🤖 OpenAI: {response}")
        except Exception as e:
            print(f"\n⚠️ Error occurred: {str(e)}")

async def cleanup(self):
    """Clean up resources."""
    await self.exit_stack.aclose()

async def main():
client = MCPClient()
try:
await client.chat_loop()
finally:
await client.cleanup()

if name == "main":
asyncio.run(main())