MCP连接静默失败求助（附报错信息与代码）

问题描述 #

• 基础MCP配置可正常运行，自定义配置连接多次失败
• 已通过uv完成环境安装与配置，MCP inspector工具运行正常
• 客户端发起连接时服务器无任何连接相关日志，报错为McpError: Connection closed

报错信息 #

(client) PS C:\Users\user\Desktop\codes\companies\Fanaye Technologies\MCP\MCP_X_Final\client> python main.py ../server/server.py
Traceback (most recent call last):
  File "C:\Users\user\Desktop\codes\companies\Fanaye Technologies\MCP\MCP_X_Final\client\main.py", line 107, in <module>
    asyncio.run(main())
  File "D:\Programs\Python\Python312\Lib\asyncio\runners.py", line 194, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "D:\Programs\Python\Python312\Lib\asyncio\runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\Programs\Python\Python312\Lib\asyncio\base_events.py", line 685, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "C:\Users\user\Desktop\codes\companies\Fanaye Technologies\MCP\MCP_X_Final\client\main.py", line 101, in main
    await mc.connect_to_server(sys.argv[1])
  File "C:\Users\user\Desktop\codes\companies\Fanaye Technologies\MCP\MCP_X_Final\client\main.py", line 30, in connect_to_server
    await self.session.initialize()
  File "C:\Users\user\Desktop\codes\companies\Fanaye Technologies\MCP\MCP_X_Final\client\.venv\Lib\site-packages\mcp\client\session.py", line 151, in initialize
    result = await self.send_request(
             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\user\Desktop\codes\companies\Fanaye Technologies\MCP\MCP_X_Final\client\.venv\Lib\site-packages\mcp\shared\session.py", line 286, in send_request    raise McpError(response_or_error.error)
mcp.shared.exceptions.McpError: Connection closed

服务器代码(server.py) #

from typing import Any
import os
import time
import asyncio
import openpyxl
import httpx
import json
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
import sys

load_dotenv()

print("Starting MCP server...", file=sys.stderr)
print(f"Python executable: {sys.executable}", file=sys.stderr)
print(f"Working directory: {os.getcwd()}", file=sys.stderr)

# Initialize FastMCP server
mcp = FastMCP("twitter_monitor")

X_API_KEY = os.getenv("X_API_KEY")
print(f"X_API_KEY configured: {bool(X_API_KEY)}", file=sys.stderr)

TWITTER_SEARCH_URL = "https://api.twitter.com/2/tweets/search/recent"
HEADERS = {
    "Authorization": f"Bearer {X_API_KEY}",
    "Content-Type": "application/json"
}


def read_excel(file_path: str, column_name: str) -> list[str]:
    print(f"Attempting to read Excel file: {file_path}", file=sys.stderr)
    
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"Excel file not found: {file_path}")
    
    try:
        wb = openpyxl.load_workbook(file_path)
        print(f"Workbook loaded. Available sheets: {wb.sheetnames}", file=sys.stderr)
    except Exception as e:
        raise Exception(f"Failed to load workbook: {e}")
    
    if "Market Makers" not in wb.sheetnames:
        raise ValueError(f"Sheet 'Market Makers' not found. Available sheets: {wb.sheetnames}")
    
    sheet = wb["Market Makers"]
    header = [cell.value for cell in next(sheet.iter_rows(min_row=1, max_row=1))]
    print(f"Excel headers: {header}", file=sys.stderr)
    
    try:
        col_idx = header.index(column_name)
    except ValueError:
        raise ValueError(f"Column '{column_name}' not found in sheet header: {header}")

    twitter_handles = set()
    for row in sheet.iter_rows(min_row=2):
        handle = row[col_idx].value
        if handle and handle not in twitter_handles:
            twitter_handles.add(handle)
            if len(twitter_handles) == 2:
                break
    
    print(f"Found Twitter handles: {list(twitter_handles)}", file=sys.stderr)
    return list(twitter_handles)


@mcp.tool()
async def get_company_tweets(column: str = "Twitter", max_results: int = 10) -> str:
    """Get tweets about Company, Announcement, or Update from Excel-provided usernames.

    Args:
        column: Column name in Excel sheet with Twitter usernames.
        max_results: Max tweets to retrieve per handle.
    """
    file_path = "mysheet.xlsx"
    try:
        twitter_handles = read_excel(file_path, column)
    except Exception as e:
        return json.dumps({"error": str(e)})

    all_posts = []
    async with httpx.AsyncClient() as client:
        for handle in twitter_handles:
            query = f"from:{handle} Company Announcement OR Update"
            params = {"query": query, "max_results": max_results}
            retries = 3

            for _ in range(retries):
                try:
                    response = await client.get(TWITTER_SEARCH_URL, headers=HEADERS, params=params)
                    
                    if response.status_code == 200:
                        all_posts.extend(response.json().get("data", []))
                        break
                    elif response.status_code == 429:
                        reset_time = int(response.headers.get("x-rate-limit-reset", 0))
                        current_time = int(time.time())
                        wait_time = max(reset_time - current_time, 5)
                        print(f"Rate limited. Waiting {wait_time}s...", file=sys.stderr)
                        await asyncio.sleep(wait_time)
                    else:
                        print(f"Error {response.status_code} for {handle}: {response.text}", file=sys.stderr)
                        break
                except Exception as e:
                    print(f"Request failed for {handle}: {e}", file=sys.stderr)
                    break

            await asyncio.sleep(1)  # Use async sleep
            if len(all_posts) >= 25:
                break

    result = {"total_fetched": len(all_posts), "tweets": all_posts[:25]}
    return json.dumps(result, indent=2)


if __name__ == "__main__":
    print("Starting MCP server...", file=sys.stderr)
    mcp.run(transport="stdio")

客户端代码(client.py) #

import os
import sys
import json
import asyncio
from contextlib import AsyncExitStack
from dotenv import load_dotenv
import httpx

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

load_dotenv()

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"

class MCPClient:
    def __init__(self):
        self.session = None
        self.exit_stack = AsyncExitStack()
        self.tool_to_session = {}
        self.deepseek_functions = []

    async def connect_to_server(self, server_script_path: str):
        is_py = server_script_path.endswith('.py')
        cmd = "python" if is_py else "node"
        params = StdioServerParameters(command=cmd, args=[server_script_path], env=None, stderr_to_stdout=True)
        read, write = await self.exit_stack.enter_async_context(stdio_client(params))
        self.session = await self.exit_stack.enter_async_context(ClientSession(read, write))
        await self.session.initialize()

        resp = await self.session.list_tools()
        for tool in resp.tools:
            self.tool_to_session[tool.name] = self.session
            self.deepseek_functions.append({
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.inputSchema
            })

        print("Connected! Tools:", [f["name"] for f in self.deepseek_functions])

    async def call_deepseek(self, messages, functions=None, function_call="auto"):
        headers = {
            "Authorization": f"Bearer {DEEPSEEK_API_KEY}",
            "Content-Type": "application/json"
        }
        body = {
            "model": "deepseek-chat",
            "messages": messages,
            "max_tokens": 1000,
            "function_call": function_call
        }
        if functions:
            body["functions"] = functions

        async with httpx.AsyncClient() as client:
            response = await client.post(DEEPSEEK_API_URL, headers=headers, json=body)
            response.raise_for_status()
            return response.json()

    async def process_query(self, query: str) -> str:
        messages = [{"role": "user", "content": query}]

        resp_json = await self.call_deepseek(messages, self.deepseek_functions)
        msg = resp_json["choices"][0]["message"]

        if "function_call" in msg:
            fname = msg["function_call"]["name"]
            fargs = json.loads(msg["function_call"]["arguments"])
            result = await self.tool_to_session[fname].call_tool(fname, fargs)

            messages.append(msg)
            messages.append({"role": "function", "name": fname, "content": result.content})

            resp2_json = await self.call_deepseek(messages)
            return resp2_json["choices"][0]["message"]["content"]

        return msg["content"]

    async def chat_loop(self):
        print("MCP Twitter Client (DeepSeek) started. Type your query or 'quit'.")
        while True:
            q = input("You: ").strip()
            if q.lower() == "quit":
                break
            try:
                print("Assistant:\n", await self.process_query(q))
            except Exception as e:
                print("Error:", e)

    async def cleanup(self):
        await self.exit_stack.aclose()

async def main():
    if len(sys.argv) != 2:
        print("Usage: python client.py <path_to_server_script>")
        sys.exit(1)
    mc = MCPClient()
    try:
        await mc.connect_to_server(sys.argv[1])
        await mc.chat_loop()
    finally:
        await mc.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

排查与修复建议 #

1. 修正Excel文件路径问题 #

服务器代码中file_path = "mysheet.xlsx"使用相对路径，而客户端启动服务器时，服务器的工作目录是客户端的当前目录，不是服务器脚本所在目录，导致文件找不到。修改为基于服务器脚本的绝对路径：

# 在server.py的get_company_tweets函数中替换file_path定义
file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "mysheet.xlsx")

2. 捕获服务器启动异常 #

在服务器入口添加异常捕获，排查启动时的崩溃问题：

if __name__ == "__main__":
    try:
        print("Starting MCP server...", file=sys.stderr)
        mcp.run(transport="stdio")
    except Exception as e:
        print(f"Server crashed on startup: {str(e)}", file=sys.stderr)
        sys.exit(1)

3. 验证服务器独立运行状态 #

单独运行服务器脚本，检查是否能正常启动：

cd path/to/server
python server.py

若启动失败，根据终端报错修复服务器自身问题。

4. 传递完整环境变量给服务器 #

客户端中StdioServerParameters的env参数设为os.environ，确保服务器能读取到环境变量：

# 在client.py的connect_to_server函数中修改params定义
params = StdioServerParameters(command=cmd, args=[server_script_path], env=os.environ, stderr_to_stdout=True)

5. 确认MCP版本兼容性 #

确保客户端与服务器使用相同版本的MCP库，版本不匹配可能导致协议不兼容：

pip show mcp

两端版本需一致，若不一致则统一升级或降级。

内容的提问来源于stack exchange，提问作者ayex