How to initialize MultiServerMCPClient as a global variable in Flask app

[gauravmindzk]

Hello everyone, I'm developing a Flask based chatbot.

In it, I'm using Langgraph's create_react_agent for question-answering and MultiServerMCPClient to get the tools provided by my MCP server.

The MCP server currently contains only 1 tool i.e. retrieve_context ( to get data from pinecone index ).

Below is the flask api route for the Ask API .

async def get_cached_client_data(client_id):
    key = cache_key(client_id)
    cached = redis_client.get(key)
    if cached:
        return json.loads(cached)
    # Not cached, fetch fresh
    auth_result = await authenticate_client(client_id)
    if not auth_result["success"]:
        return {"error": auth_result["error"]}
    token = auth_result["token"]
    config_result = await get_client_config(token)
    if not config_result["success"]:
        return {"error": config_result["error"]}
    config = config_result["config"]
    prompt_result = await get_system_prompt(token, "experienceagent")
    prompt = prompt_result["prompt"] if prompt_result["success"] else ""
    cache_data = {
        "token": token,
        "config": config,
        "prompt": prompt
    }
    redis_client.setex(key, CACHE_TTL, json.dumps(cache_data))
 
    # --- MongoClient logic tied to cache refresh ---
    # Always decode the URI if present
    mongo_uri = config.get("databaseUri") or DEFAULT_MONGO_URI
    if config.get("databaseUri"):
        decode_result = decode_base64_safely(config["databaseUri"], "databaseUri")
        if decode_result["success"]:
            mongo_uri = decode_result["decoded"]
    if mongo_uri not in mongo_clients:
        mongo_clients[mongo_uri] = MongoClient(mongo_uri)
 
    return cache_data
    
    
@app.route("/ask", methods=["POST"])
@async_route
async def ask():
    try:
        data = request.json
        prompt = data.get("prompt")
        thread_id = data.get("thread_id")
        user_id = data.get("user_id")
        client_id = data.get("client_id")

        # Validate required fields
        missing_keys = [k for k in ["prompt", "user_id", "client_id"] if not data.get(k)]
        if missing_keys:
            return jsonify({"error": f"Missing required fields: {', '.join(missing_keys)}"}), 400

        # Step 1: Get cached client data (token, config, prompt)
        client_data = await get_cached_client_data(client_id)
        if "error" in client_data:
            return jsonify({"error": client_data["error"]}), 401

        bearer_token = client_data["token"]
        client_config = client_data["config"]
        agent_name = "experienceagent"
        client_system_prompt = client_data["prompt"]

        # Step 2: Decode database URI if present
        mongodb_uri = client_config.get("databaseUri") or DEFAULT_MONGO_URI
        if client_config.get("databaseUri"):
            decode_result = decode_base64_safely(client_config["databaseUri"], "databaseUri")
            if decode_result["success"]:
                mongodb_uri = decode_result["decoded"]
            else:
                print(f"Warning: {decode_result['error']}")

        # Step 3: Use client database name if present
        db_name = client_config.get("databaseName") or "your_db_name"

        # Step 4: Initialize MongoDB client for sessions
        try:
            mongo_client = mongo_clients[mongodb_uri]
            mongo_db = mongo_client[db_name]
        except Exception as mongo_error:
            return jsonify({"error": f"Failed to connect to MongoDB: {str(mongo_error)}"}), 500

        # Step 5: Create a new thread_id if none is provided
        if not thread_id:
            try:
                result = mongo_db.sessions.insert_one({
                    "session_name": prompt,
                    "user_id": user_id,
                    "client_id": client_id
                })
                thread_id = str(result.inserted_id)
            except Exception as session_error:
                return jsonify({"error": f"Failed to create session: {str(session_error)}"}), 500

        # Step 6: Initialize OpenAI model
        if not client_config.get("apiKey"):
            return jsonify({"error": "API key not found in client configuration"}), 500

        decode_result = decode_base64_safely(client_config["apiKey"], "apiKey")
        if not decode_result["success"]:
            return jsonify({"error": decode_result["error"]}), 500

        decoded_api_key = decode_result["decoded"]
        model_name = client_config.get("modelName") or "gpt-3.5-turbo"

        try:
            model = ChatOpenAI(model=model_name, openai_api_key=decoded_api_key)
        except Exception as model_error:
            return jsonify({"error": f"Failed to initialize OpenAI model: {str(model_error)}"}), 500

        # Step 7: Using async context managers for MongoDB and MCP client
        try:
            async with AsyncMongoDBSaver.from_conn_string(mongodb_uri, db_name) as checkpointer:
                async with MultiServerMCPClient(
                    {
                        "pinecone_assistant": {
                            "url": MCP_ENDPOINT,
                            "transport": "sse"
                        }
                    }
                ) as client:
                    # Get tools from MCP client
                    try:
                        tools = client.get_tools()
                    except Exception as tools_error:
                        return jsonify({"error": f"Tool loading failed: {str(tools_error)}"}), 500

                    # Create the agent
                    try:
                        agent = create_react_agent(model, tools, prompt=client_system_prompt, checkpointer=checkpointer, name=agent_name)
                    except Exception as agent_error:
                        return jsonify({"error": f"Agent creation failed: {str(agent_error)}"}), 500

                    # Invoke the agent
                    config = {"configurable": {"thread_id": thread_id, "user_id": user_id, "client_id": client_id}}
                    try:
                        response = await agent.ainvoke({"messages": prompt}, config)
                        message = response["messages"][-1].content
                        return jsonify({"response": message, "thread_id": thread_id}), 200
                    except Exception as invoke_error:
                        return jsonify({"error": f"Agent invocation failed: {str(invoke_error)}"}), 500

        except Exception as context_error:
            return jsonify({"error": f"Context manager error: {str(context_error)}"}), 500

    except Exception as e:
        import traceback
        import ast
        import re

        tb = traceback.format_exc()
        print(f"Unexpected error in /ask endpoint: {tb}")

        error_obj = {"error": {"message": str(e), "type": "unknown_error", "param": None, "code": None}}

        try:
            match = re.search(r"\{.*'error':.*\}", str(e))
            if match:
                extracted_dict = ast.literal_eval(match.group())
                if 'error' in extracted_dict:
                    error_obj = {"error": extracted_dict['error']}
        except Exception:
            pass

        return jsonify({
            "error": error_obj,
            "traceback": tb
        }), 500

I want to initialize the 'MultiServerMCPClient' globally as it's used in my other apis as well and if possible I also want to cache it like I've cached client data