Description
Describe the bug
I 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.
In 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.
To Reproduce
Steps to reproduce the behavior:
- Deploy the FastAPI application with the MCP client and MCP server on Railway.
- Use the LLM response with arguments and the tool name.
- Observe that the tool call works on the local environment but fails in the live environment.
Expected behavior
The 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.
Screenshots
Desktop (please complete the following information):
- OS: Windows 10
- Browser chrome
- Version 136.0.7103.93
Additional context
- The tool call works locally, but not in the live environment.
- Both environments use the same configuration, but there seems to be a difference in how the SSE protocol behaves between local and live setups.
here is my code mcp client+fast api
from contextlib import asynccontextmanager
from fastapi import FastAPI
from pydantic import BaseModel
from config import SERVER_URL
from mcp_client import McpClient
from fastapi.middleware.cors import CORSMiddleware
from fastapi import HTTPException
from mcp import ClientSession
from mcp.client.sse import sse_client
class QuerySchema(BaseModel):
query:str
@asynccontextmanager
async def lifespan(app:FastAPI):
client = McpClient()
try:
connected = await client.connect_to_server('http://127.0.0.1:3000/sse')
# connected = await client.connect_to_server(f'{SERVER_URL}')
if not connected:
raise HTTPException(
status_code=500, detail="Failed to connect to MCP server"
)
app.state.client = client
print("app start")
# print(app.state.client)
yield
except Exception as e:
raise HTTPException(status_code=500, detail="Error during lifespan") from e
finally:
# shutdown when app close
print("app close")
await client.clenup()
app = FastAPI(title="MCP CLIENT CALCULATOR",lifespan=lifespan)
Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=[""], # Allows all origins
allow_credentials=True,
allow_methods=[""], # Allows all methods
allow_headers=["*"], # Allows all headers
)
@app.post('/query')
async def caluclator(request:QuerySchema):
try:
response = await app.state.client.process_query(request.query)
# print (response.status_code)
print ("response_tool",response)
return response
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error {e}") from e
if name == "main":
import uvicorn
uvicorn.run(app,host='0.0.0.0',port=8000)
mcp_client.py file
from fastapi import HTTPException
from mcp import ClientSession
from mcp.client.sse import sse_client
from typing import Optional
from contextlib import AsyncExitStack
from google import genai
from config import GOOGLE_API_KEY
from google.genai import types
class McpClient:
def init(self):
self.tools = []
self.messages = []
self.session:Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
self.client = genai.Client(api_key=GOOGLE_API_KEY)
async def connect_to_server(self,path:str):
try:
read,write = await self.exit_stack.enter_async_context(sse_client(url=path))
self.session = await self.exit_stack.enter_async_context(ClientSession(read,write))
await self.session.initialize()
# get all the tools
# get all the tools
self.tools = await self.get_mcp_tools()
print(self.tools)
return True
except Exception as e:
raise
async def get_mcp_tools(self):
try:
# print(awaitself.session.list_tools())
response = await self.session.list_tools()
return response.tools
except Exception as e:
raise(e)
async def process_query(self,query:str):
try:
""" when user hit the query save in self.messages"""
self.messages = [types.Content(role="user",parts=[types.Part(text=query)])]
# print()
# print("first message with query",self.messages)
# print()
while True:
"""now call the call_llm function to get response from gemini"""
response = await self.call_llm()
# print("response--->>>",response)
content = response.candidates[0].content
parts = content.parts
# print()
# print("second message with fnction call by llm",self.messages)
# print()
# if response.candidates[0].content.parts[0].function_call:
if parts and parts[0].function_call:
"""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"""
self.messages.append(
types.Content(role=response.candidates[0].content.role,parts=[types.Part(function_call=parts[0].function_call)])
)
function_call = parts[0].function_call
print(f"Function to call: {function_call.name}")
print(f"Arguments: {function_call.args}")
try:
"""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 """
result = await self.session.call_tool(name=function_call.name,arguments=function_call.args)
self.messages.append(
types.Content(role="tool_use",
parts=[types.Part.from_function_response(
name=function_call.name,
response={"result":result}
)]),
)
# print()
# print("third message from mcp server tool give by llm",self.messages)
"""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"""
# print()
final_result = self.client.models.generate_content(
model="gemini-2.5-flash-preview-04-17",
config=types.GenerateContentConfig(temperature=1),
contents=self.messages
)
self.messages.append(
types.Content(role=final_result.candidates[0].content.role,parts=[types.Part(text=final_result.candidates[0].content.parts[0].text)])
)
break
except Exception as e:
# print(f"error get tool response of {function_call.name}")
# print(f"error get tool response of {function_call.args}")
raise HTTPException(status_code=500,details = f"Error calling tool {e}")
else:
print("No function call found in the response.")
print(response.candidates[0].content.parts[0].text)
self.messages.append(
types.Content(
role=content.role,
parts=[types.Part(text=response.candidates[0].content.parts[0].text)]
)
)
break
return self.messages
except Exception as e:
raise(e)
async def call_llm(self):
"""Call llm with tools get when app start first time and with self.messages having user quer and user role"""
try:
available_tools = [
types.Tool(
function_declarations=[
{
"name": tool.name,
"description": tool.description,
"parameters": {
k: v
for k, v in tool.inputSchema.items()
if k not in ["additionalProperties", "$schema"]
},
}
]
)
for tool in self.tools
]
# print(available_tools)
return self.client.models.generate_content(
model="gemini-2.5-flash-preview-04-17",
config=types.GenerateContentConfig(temperature=1,tools=available_tools),
contents=self.messages
)
except Exception as e:
raise(e)
async def call_tool(self,arguments:dict):
try:
# print(a,b,operator)
response = await self.session.call_tool(name="mcpCalculator",arguments=arguments)
return response
except Exception as e:
raise(e)
async def clenup(self):
try:
await self.exit_stack.aclose()
print("Disconnected from MCP server")
except Exception as e:
raise(e)