René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Error when calling MCP tool with arguments and tool name in live environment","articleBody":"**Describe the bug**\nI am experiencing an issue when calling the MCP tool with arguments and tool name in the live environment using Railway. I am hosting both the FastAPI with the MCP client and the MCP server on Railway. The tool call works fine in the local environment, but when deployed on Railway (live), the tool does not work as expected. The issue arises after the LLM response with arguments and the tool name.\n\nIn the local environment, I use the SSE protocol to interact with the MCP server and everything works fine. However, in the live environment, the tool call fails despite receiving all tools correctly from the MCP server when the app starts.\n\n**To Reproduce**\nSteps to reproduce the behavior:\n1. Deploy the FastAPI application with the MCP client and MCP server on Railway.\n2. Use the LLM response with arguments and the tool name.\n3. Observe that the tool call works on the local environment but fails in the live environment.\n\n**Expected behavior**\nThe tool call should work correctly both in the local environment and live environment. It should successfully invoke the tool when passing the correct arguments and tool name.\n\n**Screenshots**\n\n![Image](https://github.com/user-attachments/assets/50a48783-f3f3-499f-b69a-763c59161e3b)\n\n**Desktop (please complete the following information):**\n - OS: Windows 10\n - Browser chrome\n - Version 136.0.7103.93\n\n**Additional context**\n- The tool call works locally, but not in the live environment.\n- Both environments use the same configuration, but there seems to be a difference in how the SSE protocol behaves between local and live setups. \n\nhere is my code mcp client+fast api\nfrom contextlib import asynccontextmanager\nfrom fastapi import FastAPI\nfrom pydantic import BaseModel\nfrom config import SERVER_URL\nfrom mcp_client import McpClient\nfrom fastapi.middleware.cors import CORSMiddleware\nfrom fastapi import HTTPException\nfrom mcp import ClientSession\nfrom mcp.client.sse import sse_client\n\nclass QuerySchema(BaseModel):\n    query:str\n\n@asynccontextmanager\nasync def lifespan(app:FastAPI):\n    client = McpClient()\n    try:\n        connected = await client.connect_to_server('http://127.0.0.1:3000/sse')\n        # connected = await client.connect_to_server(f'{SERVER_URL}')\n        if not connected:\n             raise HTTPException(\n                status_code=500, detail=\"Failed to connect to MCP server\"\n            )\n        app.state.client = client\n        print(\"app start\")\n        # print(app.state.client)\n        yield \n    except Exception as e:\n        raise HTTPException(status_code=500, detail=\"Error during lifespan\") from e\n    finally:\n        # shutdown when app close\n        print(\"app close\")\n        await client.clenup()\n\napp = FastAPI(title=\"MCP CLIENT CALCULATOR\",lifespan=lifespan)\n\n\n# Add CORS middleware\napp.add_middleware(\n    CORSMiddleware,\n    allow_origins=[\"*\"],  # Allows all origins\n    allow_credentials=True,\n    allow_methods=[\"*\"],  # Allows all methods\n    allow_headers=[\"*\"],  # Allows all headers\n)\n\n\n\n@app.post('/query')\nasync def caluclator(request:QuerySchema):\n    try:\n        response  = await app.state.client.process_query(request.query)\n        # print (response.status_code)\n        print (\"response_tool\",response)\n        return response\n\n    except Exception as e:\n        raise HTTPException(status_code=500, detail=f\"Error {e}\") from e\n        \n\nif __name__ == \"__main__\":\n    import uvicorn    \n    uvicorn.run(app,host='0.0.0.0',port=8000)\n\nmcp_client.py file\nfrom fastapi import HTTPException\nfrom mcp import ClientSession\nfrom mcp.client.sse import sse_client\nfrom typing import Optional\nfrom contextlib import AsyncExitStack\nfrom google import genai\nfrom config import GOOGLE_API_KEY\nfrom google.genai import types\nclass McpClient:\n    def __init__(self):\n        self.tools = []\n        self.messages = []\n        self.session:Optional[ClientSession]  = None\n        self.exit_stack = AsyncExitStack()\n        self.client = genai.Client(api_key=GOOGLE_API_KEY)\n        \n    async def connect_to_server(self,path:str):\n        try:\n            read,write = await self.exit_stack.enter_async_context(sse_client(url=path))\n            self.session = await self.exit_stack.enter_async_context(ClientSession(read,write))\n            await self.session.initialize()\n            # get all the tools\n            # get all the tools\n            self.tools = await self.get_mcp_tools()\n            print(self.tools)\n            return True\n                \n        except Exception as e:\n              raise\n        \n    async def get_mcp_tools(self):\n        try:\n            # print(awaitself.session.list_tools())\n            response = await self.session.list_tools()\n            return response.tools\n        except Exception as e:\n             raise(e)      \n    \n    async def process_query(self,query:str):\n        try:\n            \"\"\" when user hit the query save in self.messages\"\"\" \n            self.messages = [types.Content(role=\"user\",parts=[types.Part(text=query)])]\n            # print()\n            # print(\"first message with query\",self.messages)\n            # print()\n            while True:\n                \"\"\"now call the call_llm function to get response from gemini\"\"\" \n                response  = await self.call_llm()\n                # print(\"response---\u003e\u003e\u003e\",response)\n                content = response.candidates[0].content\n                parts = content.parts\n                \n                # print()\n                # print(\"second message with fnction call by llm\",self.messages)\n                # print()\n                \n                # if response.candidates[0].content.parts[0].function_call:\n                if parts and parts[0].function_call:\n                    \"\"\"After call call_llm llm return the function tool that we send with useer query and all tools fetch from mcp server and now append with self.messages funtion call send by llm according to query\"\"\"\n                    self.messages.append(\n                    types.Content(role=response.candidates[0].content.role,parts=[types.Part(function_call=parts[0].function_call)])\n                    )\n                    \n                    function_call = parts[0].function_call\n                    print(f\"Function to call: {function_call.name}\")\n                    print(f\"Arguments: {function_call.args}\")\n                    try:\n                        \"\"\"After get tool send by llm now call that tool to mcp server to fetch data according to that function and append result to self.messages \"\"\"\n                        result = await self.session.call_tool(name=function_call.name,arguments=function_call.args)\n                        \n                        self.messages.append(\n                            types.Content(role=\"tool_use\",\n                            parts=[types.Part.from_function_response(\n                                name=function_call.name,\n                                response={\"result\":result}\n                            )]),\n                            \n                        )\n                        # print()\n                        # print(\"third message from mcp server tool give by llm\",self.messages)\n                        \"\"\"now finally send this self.messages list to llm to get final result that get from mcp server by calling tool purpose of this call to llm to get more understandable answer\"\"\"\n                        # print()\n                        final_result = self.client.models.generate_content(\n                            model=\"gemini-2.5-flash-preview-04-17\",\n                            config=types.GenerateContentConfig(temperature=1),\n                            contents=self.messages\n                        )\n                        self.messages.append(\n                            types.Content(role=final_result.candidates[0].content.role,parts=[types.Part(text=final_result.candidates[0].content.parts[0].text)]) \n                        )\n                        break\n                    except Exception as e:\n                        # print(f\"error get tool response of {function_call.name}\")\n                        # print(f\"error get tool response of {function_call.args}\")\n                        raise HTTPException(status_code=500,details = f\"Error calling tool {e}\")\n                else:\n                    print(\"No function call found in the response.\")\n                    print(response.candidates[0].content.parts[0].text)\n                    self.messages.append(\n                            types.Content(\n                                role=content.role,\n                                parts=[types.Part(text=response.candidates[0].content.parts[0].text)]\n                            )\n                        \n                    )\n                    break\n            return self.messages\n                    \n        except Exception as e:\n            raise(e)\n\n    async def call_llm(self):\n        \"\"\"Call llm with tools get when app start first time and with self.messages having user quer and user role\"\"\"\n        try:\n            available_tools = [\n                types.Tool(\n                    function_declarations=[\n                        {\n                            \"name\": tool.name,\n                            \"description\": tool.description,\n                            \"parameters\": {\n                                k: v\n                                for k, v in tool.inputSchema.items()\n                                if k not in [\"additionalProperties\", \"$schema\"]\n                            },\n                        }\n                    ]\n                )\n                for tool in self.tools\n            ]\n            # print(available_tools)\n            return self.client.models.generate_content(\n                    model=\"gemini-2.5-flash-preview-04-17\",\n                    config=types.GenerateContentConfig(temperature=1,tools=available_tools),\n                    contents=self.messages\n                )\n        except Exception as e:\n            raise(e)\n\n    async def call_tool(self,arguments:dict):\n        try:\n            # print(a,b,operator)\n            response  = await self.session.call_tool(name=\"mcpCalculator\",arguments=arguments)\n            return response\n\n        except Exception as e:\n             raise(e)\n    \n    async def clenup(self):\n        try:\n            await self.exit_stack.aclose()\n            print(\"Disconnected from MCP server\")\n        except Exception as e:\n            raise(e)\n","author":{"url":"https://github.com/Muhammadtalha678","@type":"Person","name":"Muhammadtalha678"},"datePublished":"2025-05-08T07:07:38.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":1},"url":"https://github.com/662/python-sdk/issues/662"}