How I can realtime update the Ui when I receive a reuest upon fast api?

Question

I have this simple script:

import os
import gradio as gr
from fastapi import FastAPI, Request
import uvicorn
import threading
from typing import List
from datetime import datetime

api = FastAPI()

# Shared logs

class Log():
    def __init__(self):
        self._logs: List[str] = []
        self.logstr=""
    
    def log_message(self,msg: str):
        timestamp = datetime.now().strftime("%H:%M:%S")
        self._logs.append(f"[{timestamp}] {msg}")
        self.logstr="\n".join(self._logs)


log = Log()
log_state=gr.State(log)

# === FastAPI Setup ===


@api.post("/log")
async def receive_log(request: Request):
    data = await request.body()
    msg = f"API received: {data}"
    log.log_message(msg)
    gr.update(value=log.logstr)
    return {"status": "logged", "message": msg}

def run_api():
    api_port = int(os.environ.get("API_PORT", 8000))
    uvicorn.run(api, host="0.0.0.0", port=api_port)

# === Gradio UI ===

with gr.Blocks() as ui:
    gr.Markdown("## 📝 Incoming HTTP Requests")
    log_box = gr.Textbox(label="Logs", inputs=log_state, lines=20)
    
    # Trigger the refresh when the log state is updated


def run_gradio():
    gradio_port = int(os.environ.get("GRADIO_PORT", 7860))
    ui.launch(server_port=gradio_port)

# === Start Both ===
if __name__ == "__main__":
    threading.Thread(target=run_api, daemon=True).start()
    run_gradio()

What I try to achieve is to have FastApi listening to one port and a panel that displays upon realtime the incomming requests:

POST /log -> FastAPI ->common_log -> Gradio

But I am unable to change the contents of Textbox when I receive Incomming requests in FastAPI. How I can do this?

Answer 1

Only method which works for me is

def function():
    return log.logstr

Textbox(value=function, ..., every=1)

It runs function every 1 second, and this function returns current content in log.

Doc: Gradio Textbox

---

Full code:

import os
import gradio as gr
from fastapi import FastAPI, Request
import uvicorn
import threading
from typing import List
from datetime import datetime

api = FastAPI()

# Shared logs

class Log():
    def __init__(self):
        self._logs: List[str] = []
        self.logstr=""

    def log_message(self,msg: str):
        timestamp = datetime.now().strftime("%H:%M:%S")
        self._logs.append(f"[{timestamp}] {msg}")
        self.logstr="\n".join(self._logs)

log = Log()
log_state = gr.State(log)

# === FastAPI Setup ===

@api.post("/log")
async def receive_log(request: Request):
    data = await request.body()
    msg = f"API received: {data}"
    log.log_message(msg)
    gr.update(value=log.logstr)
    #print('data:', data)
    return {"status": "logged", "message": msg}

def run_api():
    #print('run FastAPI')
    api_port = int(os.environ.get("API_PORT", 8000))
    uvicorn.run(api, host="0.0.0.0", port=api_port)

# === Gradio UI ===

def get_logs():
    #print('run: get_logs')
    return log.logstr

with gr.Blocks() as ui:
    gr.Markdown("## 📝 Incoming HTTP Requests")
    log_box = gr.Textbox(label="Logs", value=get_logs, lines=20, every=1)
    # Trigger the refresh when the log state is updated

def run_gradio():
    gradio_port = int(os.environ.get("GRADIO_PORT", 7860))
    ui.launch(server_port=gradio_port)

# === Start Both ===

if __name__ == "__main__":
    threading.Thread(target=run_api, daemon=True).start()
    run_gradio()