3

I have deployed a python script for mcp server in a docker container on Google Cloud Run. Below is a sample script

import asyncio
import logging
import os

from fastmcp import FastMCP 

logger = logging.getLogger(__name__)
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)

mcp = FastMCP("MCP Server on Cloud Run")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Use this to add two numbers together.
    
    Args:
        a: The first number.
        b: The second number.
    
    Returns:
        The sum of the two numbers.
    """
    logger.info(f">>> Tool: 'add' called with numbers '{a}' and '{b}'")
    return a + b

if __name__ == "__main__":
    logger.info(f" MCP server started on port {os.getenv('PORT', 8080)}")
    # Could also use 'sse' transport, host="0.0.0.0" required for Cloud Run.
    asyncio.run(
        mcp.run_async(
            transport="streamable-http", 
            host="0.0.0.0", 
            port=os.getenv("PORT", 8080),
        )
    ) 

I have put this in docker and deployed the image to CloudRun and got the https endpoint for calling a streamable https request. The Cloud Run service showing the deployment

I have created a Service account with Cloud Run Invoker permission and generated a json key. But when I try to access the service from python I am getting 403 unauthorised error. I used the below code to try to call the mcp server.

import os
import json
import requests
import google.oauth2.id_token
import google.auth.transport.requests

def runCloudFunction():
    os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = 'path\to\file.json'
    request = google.auth.transport.requests.Request()
    audience = 'https://cloud_run_service_url'
    TOKEN = google.oauth2.id_token.fetch_id_token(request, audience)
    print(TOKEN)
    r=requests.post(
        audience+'/mcp',
        headers={'Authorization':"Bearer "+TOKEN, 'Content-Type':'application/json'})
    print(r.status_code)

if __name__ == "__main__":
    runCloudFunction()    

The above code is printing the token but returning status 403 for the request to the service. I do not want to remove authentication from the service since that will make it insecure.So , I have selected the Require Authentication option. Security Settings for Cloud Service

I checked that public access is enabled for the Cloud Service in Networking Settings. Networking Settings

Will be really grateful if someone can let me know what I missed. I am not aware of the body to pass to the service to call a particular python function/mcp-tool. WIll be helpful if someone can guide on that as well. Thank you in advance.

3
  • 1
    What happens with url=f"{audience}/mcp/"? The URL includes a terminating "/" Commented Jul 1 at 21:56
  • You're going to have problems with you test client too since it doesn't implement the MCP protocol. You should consider using fastmcp.Client which you can configure to provide the bearer token Commented Jul 1 at 22:54
  • Consider using export GOOGLE_APPLICATION_CREDENTIALS=/path/to/file.json and letting ADC in your code find these credentials rather than the static binding that you have (this is an anti-pattern). Commented Jul 1 at 22:55

1 Answer 1

1

The following works for me.

In reverse order:

export GOOGLE_APPLICATION_CREDENTIALS=${PWD}/tester.json
export CLOUD_RUN_URL=$(\
  gcloud run services describe ${NAME} \
  --region=${REGION} \
  --project=${PROJECT} \
  --format="value(status.url)")

uv run add.py 25 17 # add(25,17)

Yields:

Connected
[TextContent(type='text', text='42', annotations=None)]

With: pyproject.toml

[project]
name = "79685701"
version = "0.0.1"
description = "Stackoverflow: 79685701"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
    "fastmcp>=2.9.2",
    "google-auth>=2.40.3",
    "requests>=2.32.4"
]

main.py:

import asyncio
import os

from fastmcp import FastMCP, Context

mcp = FastMCP("MCP Server on Cloud Run")

@mcp.tool()
async def add(a: int, b: int, ctx: Context) -> int:
    await ctx.debug(f"[add] {a}+{b}")
    result = a+b
    await ctx.debug(f"result={result}")
    return result


if __name__ == "__main__":
    asyncio.run(
        mcp.run_async(
            transport="streamable-http", 
            host="0.0.0.0", 
            port=os.getenv("PORT", 8080),
        )
    )

add.py:

from fastmcp import Client

import asyncio
import google.oauth2.id_token
import google.auth.transport.requests
import os
import sys

args = sys.argv
if len(args) != 3:
    sys.stderr.write(f"Usage: python {args[0]} <a> <b>\n")
    sys.exit(1)

a = args[1]
b = args[2]

audience = os.getenv("CLOUD_RUN_URL")

request = google.auth.transport.requests.Request()
token = google.oauth2.id_token.fetch_id_token(request, audience)

config = {
    "mcpServers": {
        "cloud-run":{
            "transport": "http",
            "url": f"{audience}/mcp/",
            "headers": {
                "Authorization": "Bearer token",
            },
            "auth": token,
        }
    }
}

client = Client(config)


async def run():
    async with client:
        print("Connected")
        result = await client.call_tool(
            name="add",
            arguments={"a":a, "b":b},
        )
        print(result)


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

Dockerfile:

# FastMCP Application Dockerfile
FROM docker.io/python:3.13-slim

RUN apt-get update && \
    apt-get install -y --no-install-recommends curl ca-certificates

ADD https://astral.sh/uv/install.sh /uv-installer.sh

RUN sh /uv-installer.sh && \
    rm /uv-installer.sh

ENV PATH="/root/.local/bin:${PATH}"

WORKDIR /app

COPY main.py main.py
COPY pyproject.toml pyproject.toml
COPY uv.lock uv.lock

RUN uv sync --locked

EXPOSE 8080

ENTRYPOINT ["uv", "run","/app/main.py"]

And:

BILLING="..."
PROJECT="..."

NAME="fastmcp"

REGION="..."

ACCOUNT="tester"
EMAIL=${ACCOUNT}@${PROJECT}.iam.gserviceaccount.com

gcloud iam service-accounts create ${ACCOUNT} \
--project=${PROJECT}

gcloud iam service-accounts keys create ${PWD}/${ACCOUNT}.json \
 --iam-account=${EMAIL} \
 --project=${PROJECT}

gcloud projects add-iam-policy-binding ${PROJECT} \
--member=serviceAccount:${EMAIL} \
--role=roles/run.invoker

gcloud auth print-access-token \
| podman login ${REGION}-docker.pkg.dev \
  --username=oauth2accesstoken \
  --password-stdin

REPO="cloud-run-source-deploy"
VERS="0.0.1"
IMAGE=${REGION}-docker.pkg.dev/${PROJECT}/${REPO}/${NAME}:${VERS}

podman build \
--tag=${IMAGE} \
--file=${PWD}/Dockerfile \
${PWD}

podman push ${IMAGE}

gcloud run deploy ${NAME} \
--image=${IMAGE} \
--region=${REGION} \
--project=${PROJECT} \
--no-allow-unauthenticated
Sign up to request clarification or add additional context in comments.

8 Comments

Wow !! Thank you for the detailed steps. Will follow this and confirm if it works.
Hi @DazWilkin I followed these steps and was able to deploy it successfully. But when I tried to call it with the client, I got the below exception. pydantic_core._pydantic_core.ValidationError: 6 validation errors for MCPConfig mcpServers.cloud-run.StdioMCPServer.command Field required [type=missing, input_value={'transport': 'http', 'ur...5HZSafzOzugcl6AxBz3BzQ'}, input_type=dict] For further information visit errors.pydantic.dev/2.11/v/missing mcpServers.cloud-run.StdioMCPServer.transport
Looks like a Pydantic error. Is the dictionary structure correct ? I tried to check with other questions but no similar examples could be found.
It worked when I changed http to streamable-http in the client. config = { "mcpServers": { "cloud-run":{ "transport": "streamable-http", "url": f"{audience}/mcp/", "headers": { "Authorization": "Bearer token", }, "auth": token, } } }
Sorry that it didn't work for you. Is your only change from "transport": "http" to "transport": "streamable-http"? Because both work for me.
|

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.