Can't get a real response from Vertex AI on Cloud Run
I'm new to Vertex AI/hosting things on Cloud Run.
I went to Prompt management, created a prompt, and clicked Build with code. Now my prompt is hosted on Google run in a Gradio app.
When I go to the page and try it, everything is fine.
When I make a POST request to the endpoint, I just get this gibberish back:
{
"data": [
{
"interactive": false,
"placeholder": "",
"value": {
"text": "",
"files": []
},
"__type__": "update"
},
null
],
"is_generating": false,
"duration": 0.0010356903076171875,
"average_duration": 0.0012654636217200239,
"render_config": null,
"changed_state_ids": []
}
AND, my logs are not showing up in the Cloud Run logger (or anywhere else I can find). All my users have the Logs Writer permission.
My Python code looks like this, I've edited it so I can get a response back from a REST call:
Dockerfile:
FROM python:3.11-slim
WORKDIR /usr/src/app
COPY . .
RUN pip install --require-hashes -r base-tooling-requirements.txt
RUN pip install --require-hashes -r requirements.txt
EXPOSE 8080
ENV GRADIO_SERVER_PORT="8080"
ENV GRADIO_SERVER_NAME="0.0.0.0"
CMD ["python", "app.py"]
app.py:
"""Main entry point for the app.
This app is generated based on your prompt in Vertex AI Studio using
Google GenAI Python SDK (https://googleapis.github.io/python-genai/) and
Gradio (https://www.gradio.app/).
You can customize the app by editing the code in Cloud Run source code editor.
You can also update the prompt in Vertex AI Studio and redeploy it.
"""
import base64
from google import genai
from google.genai import types
import gradio as gr
import utils
import logging
logging.basicConfig(level=logging.INFO)
def main():
logging.info("THIS IS INFO")
logging.warning("THIS IS WARNING")
logging.error("THIS IS ERROR")
raise gr.Error("this is error", None, title="this is error")
if __name__ == "__main__":
main()
logger.info("got here")
def generate(
message,
request: gr.Request
):
"""Function to call the model based on the request."""
logging.info("got here 2")
validate_key_result = utils.validate_key(request)
if validate_key_result is not None:
yield validate_key_result
return
client = genai.Client(
vertexai=True,
project="no-bs-czech",
location="europe-central2",
)
model = "gemini-2.5-flash"
prompt_part = types.Part.from_text(
text="Correct the Czech text I send you for spelling, grammar, syntax, and wording, in an informal tone, and send back the corrected text. Explain your corrections in English."
)
logging.info(message)
message_text = ""
if isinstance(message, dict):
message_text = message.get("text", "")
else:
message_text = message
user_part = types.Part.from_text(text=message_text)
logging.info(message_text)
contents = [
types.Content(role="user", parts=[prompt_part]),
types.Content(role="user", parts=[user_part])
]
generate_content_config = types.GenerateContentConfig(
temperature=1,
top_p=1,
seed=0,
max_output_tokens=65535,
stop_sequences=["-!"],
safety_settings=[
types.SafetySetting(
category="HARM_CATEGORY_HATE_SPEECH",
threshold="OFF"
),
types.SafetySetting(
category="HARM_CATEGORY_DANGEROUS_CONTENT",
threshold="OFF"
),
types.SafetySetting(
category="HARM_CATEGORY_SEXUALLY_EXPLICIT",
threshold="OFF"
),
types.SafetySetting(
category="HARM_CATEGORY_HARASSMENT",
threshold="OFF"
)
],
response_mime_type="application/json",
response_schema={"type":"OBJECT","properties":{"original_text":{"type":"STRING"},"updated_text":{"type":"STRING"},"changes":{"type":"ARRAY","items":{"type":"OBJECT","properties":{"original_text":{"type":"STRING"},"updated_text":{"type":"STRING"},"reason":{"type":"STRING"}}}}}},
)
response = client.models.generate_content(
model=model,
contents=contents,
config=generate_content_config,
)
logging.info(response)
text = ""
try:
# If your response is a dict with a 'candidates' list and 'content'
text = response.candidates[0].content.parts[0].text
except Exception:
text = str(response)
logging.info(text)
return {"text": text, "files": []}
iface = gr.Interface(
fn=generate,
inputs=gr.MultimodalTextbox(),
outputs=gr.MultimodalTextbox(),
title="Czech Text Correction and Explanation - Informal"
)
logging.info("iface")
# Launch the API endpoint (exposes /api/predict/)
iface.launch(show_error=True)
utils.py:
"""Utils for the app."""
import base64
import io
import mimetypes
import typing
from google.genai import types
import gradio as gr
from PIL import Image
# Error message for invalid key.
ker_error_msg = """Please open the app from
Vertex AI Studio
, not Cloud Run.
Or, obtain the key from the "Manage App" dialog within Vertex AI Studio and
append it to the url as "?key=SECRET_KEY"."""
# Google's Blue color theme
google_blue_color_hue = gr.themes.Color(
name="google_blue",
c50="#E8F0FE",
c100="#D2E3FC",
c200="#AECBFA",
c300="#8AB4F8",
c400="#669DF6",
c500="#4285F4",
c600="#1A73E8",
c700="#1967D2",
c800="#185ABC",
c900="#0f172a",
c950="#174EA6",
)
# Custom theme for the app.
custom_theme = gr.themes.Default(
primary_hue=google_blue_color_hue,
secondary_hue=google_blue_color_hue,
font=[gr.themes.GoogleFont("Google Sans")]
).set(
button_cancel_background_fill="*secondary_200",
button_cancel_background_fill_dark="*secondary_200",
button_cancel_background_fill_hover="*secondary_300",
button_cancel_background_fill_hover_dark="*secondary_300",
button_cancel_text_color="black",
button_cancel_text_color_dark="white",
)
public_access_warning = """
⚠️
Warning: This app allows unauthenticated access by default. Avoid using it for sensitive data. Access control is coming soon.
"""
next_steps_html = """
Next steps:
- Go to Cloud Run
source code editor
to customize application code.
- Go to your
Vertex AI Studio prompt
to update and redeploy it.
- Go to Cloud Run
Security settings
to turn off unauthenticated access when it's not needed.
Learn more
"""
def validate_key(request):
"""Help function to validate the key.
Args:
request: The request object.
Returns:
None if the key is valid, otherwise an error.
"""
secret_key = "SECRET_KEY"
if not secret_key:
return None
error_title = None
key = request.query_params.get("key", None)
if key is None:
error_title = "[Authorization error] No secret key provided in the URL"
elif key != secret_key:
error_title = (
f"""[Authorization error] The provided key ("{key}") is invalid."""
)
if error_title is not None:
raise gr.Error(ker_error_msg, None, title=error_title)
def get_part_from_file(file):
"""Help function to get the part from a file."""
guessed_type = mimetypes.guess_type(file)
if guessed_type:
mime_type = guessed_type[0]
else:
mime_type = "application/octet-stream"
with open(file, "rb") as f:
data = f.read()
return types.Part.from_bytes(
data=data,
mime_type=mime_type,
)
def get_bytes_from_image(image: Image.Image, mime_type: str = "PNG") -> bytes:
"""Converts a PIL Image object to bytes in the specified format.
Args:
image: The PIL Image object.
mime_type: The image format to save as (e.g., 'PNG', 'JPEG', 'GIF').
Defaults to 'PNG'.
Returns:
A bytes object representing the image in the specified format.
"""
img_byte_arr = io.BytesIO()
image.save(img_byte_arr, format=mime_type)
img_byte_arr = img_byte_arr.getvalue()
return img_byte_arr
def get_parts_from_message(
message: typing.Union[str, tuple[str, ...], dict[str, str], gr.Image],
):
"""Help function to get the parts from a message."""
parts = []
if isinstance(message, dict):
parts = []
if "text" in message and message["text"]:
parts.append(types.Part.from_text(text=message["text"]))
if "files" in message:
for file in message["files"]:
parts.append(get_part_from_file(file))
elif isinstance(message, str):
if message:
parts.append(types.Part.from_text(text=message))
elif isinstance(message, gr.Image):
if message.type == "pil":
bytes_data = get_bytes_from_image(message.value)
parts.append(
types.Part.from_bytes(data=bytes_data, mime_type=message.format)
)
elif message.type == "filepath":
parts.append(get_part_from_file(message.value))
else:
for part in list(message):
if part.startswith("/tmp/gradio"):
parts.append(get_part_from_file(part))
elif part:
parts.append(types.Part.from_text(text=part))
# To avoid error when sending empty message.
if not parts:
parts.append(types.Part.from_text(text=" "))
return parts
def convert_blob_to_gr_image(blob: types.Blob) -> gr.Image:
"""Converts a blob of image data to a gr.Image object."""
blob_data = blob.data
# Create an in-memory binary stream using io.BytesIO
image_stream = io.BytesIO(blob_data)
# Open the image from the stream using PIL.Image.open()
image = Image.open(image_stream)
return gr.Image(image)
def image_blob_to_markdown_base64(blob: types.Blob) -> str:
"""Converts image bytes to a Markdown displayable string using Base64 encoding."""
blob_data = blob.data
base64_string = base64.b64encode(blob_data).decode("utf-8")
markdown_string = (
f'
'
)
return markdown_string
def convert_part_to_gr_type(
part: types.Part,
use_markdown: bool = False,
) -> typing.Optional[typing.Union[str, gr.Image]]:
"""Converts a part object to a str or gr.Image object."""
if part.text:
return part.text
elif part.inline_data:
if use_markdown:
return image_blob_to_markdown_base64(part.inline_data)
return convert_blob_to_gr_image(part.inline_data)
else:
return None
def convert_content_to_gr_type(
content: typing.Optional[types.Content],
use_markdown: bool = False,
) -> typing.Optional[typing.Union[str, gr.Image]]:
"""Converts a content object to a gr.ChatMessage object."""
if content is None or content.parts is None:
return []
results = [
convert_part_to_gr_type(part, use_markdown) for part in content.parts
]
return [res for res in results if res is not None]
I'm trying to make a POST request to the endpoint (in Postman):
body:
{
"data": [
{ "text": "dobry den", "files": [] }
],
"fn_index": 0
}
How can I get the response back from the request to the model?