Implementing MCP Client via XXXCloud API

pip install mcp
pip install openai

import asyncio
import json
from typing import Optional
from contextlib import AsyncExitStack
 
from openai import OpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
 
 
openai_client = OpenAI(
    base_url="https://api.umodelverse.ai/v1",  # Modelverse API URL (do not change)
    api_key="<Your_api_key>",  # API key created in the console
)
model_name = "Qwen/QwQ-32B"  # Model name (do not change)
 
 
class MCPClient:
    def __init__(self):
        """Initialize MCP client"""
        self.exit_stack = AsyncExitStack()
        self.client = openai_client
        # Initialize client
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
 
    async def connect_to_server(self):
        # Using a keyless MCP Server
        server_params = StdioServerParameters(
            # Command for macOS
            command="npx",
            args=["-y", "@mzxrai/mcp-webresearch@latest"],
            # Command for Windows
            # command="cmd.exe", args=["/c", "npx", "-y", "@mzxrai/mcp-webresearch@latest"],
        )
 
        # Start MCP server and establish communication
        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()
 
        # List tools on the MCP server
        list_tools_resp = await self.session.list_tools()
        list_prompt_resp = await self.session.list_prompts()
        list_resource_resp = await self.session.list_resources()
        self.session.get_prompt
        tools = list_tools_resp.tools
        print(
            "\nConnected to server with the following tools:",
            [tool.name for tool in tools],
            "Available prompts:",
            [prompt.name for prompt in list_prompt_resp.prompts],
            "Available resources:",
            [resource.name for resource in list_resource_resp.resources],
        )
 
    async def process_prompt(self, query: str) -> str:
        """
        Process queries using the large model and call available MCP tools (Tool Calling)
        """
        messages = [{"role": "user", "content": query}]
 
        list_tools_resp = await self.session.list_tools()
        self.session.complete
 
        available_tools = [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema,
                },
            }
            for tool in list_tools_resp.tools
        ]
 
        resp = self.client.chat.completions.create(
            model=model_name, messages=messages, tools=available_tools
        )
        # Process the returned content
        content = resp.choices[0]
        if content.finish_reason == "tool_calls":
            # Parse the tool to be used if needed
            tool_call = content.message.tool_calls[0]
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments)
 
            # Execute tool
            result = await self.session.call_tool(tool_name, tool_args)
            print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n")
 
            # Store both the data returned from the model indicating which tool to call and the data after tool execution in messages
            messages.append(
                {
                    "role": "tool",
                    "content": result.content[0].text,
                    "tool_call_id": tool_call.id,
                }
            )
 
            # Return the results above to the large model for generating the final outcome
            resp = self.client.chat.completions.create(
                messages=messages, model=model_name, tools=available_tools
            )
            return resp.choices[0].message.content
 
        return content.message.content
 
    async def chat_loop(self):
        """Run an interactive chat loop"""
        print("\n🤖 MCP Host is running! Type 'exit' to exit")
 
        while True:
            try:
                prompt = input("\nYou: ").strip()
                if prompt.lower() == "exit":
                    break
 
                response = await self.process_prompt(prompt)
                print(f"\n🤖 ModelVerse QwQ-32B: {response}")
 
            except Exception as e:
                print(f"\n⚠️ An error occurred: {str(e)}")
 
    async def cleanup(self):
        """Clean up resources"""
        await self.exit_stack.aclose()
 
 
async def main():
    client = MCPClient()
    try:
        await client.connect_to_server()
        await client.chat_loop()
    finally:
        await client.cleanup()
 
 
if __name__ == "__main__":
 
    asyncio.run(main())