diff --git a/alloydb-ai-mcp/cymbal_logistic/.gitignore b/alloydb-ai-mcp/cymbal_logistic/.gitignore new file mode 100644 index 0000000..f9cd96a --- /dev/null +++ b/alloydb-ai-mcp/cymbal_logistic/.gitignore @@ -0,0 +1 @@ +internal/ diff --git a/alloydb-ai-mcp/cymbal_logistic/README.md b/alloydb-ai-mcp/cymbal_logistic/README.md index 3797761..5ec7921 100644 --- a/alloydb-ai-mcp/cymbal_logistic/README.md +++ b/alloydb-ai-mcp/cymbal_logistic/README.md @@ -1,4 +1,4 @@ -# Cymbal Logistic Agent +# Cymbal Logistics Agent This application is an AI-powered conversational agent that leverages the Google Autonomous Data Kit (ADK) alongside a sleek frontend generated with Google Mesop. diff --git a/alloydb-ai-mcp/cymbal_logistic/app.py b/alloydb-ai-mcp/cymbal_logistic/app.py index 9372807..4890b67 100644 --- a/alloydb-ai-mcp/cymbal_logistic/app.py +++ b/alloydb-ai-mcp/cymbal_logistic/app.py @@ -57,9 +57,15 @@ class State: has_chart: bool = False debug_info: str = "" enable_debug: bool = False + query_history_json: str = "[]" + active_query_tab: int = 0 session_id: str = "" conversation_history: list[str] = field(default_factory=list) context_summary: str = "" + +def select_tab(e: me.ClickEvent): + state = me.state(State) + state.active_query_tab = int(e.key) def on_cluster_name_change(e: me.InputBlurEvent): state = me.state(State) @@ -98,7 +104,6 @@ def __init__(self): # We create a new runner instance per request to avoid session locking bugs def run_query_sync(request_text, cluster_name, location, instance_name, database_name, project_id, session_id, summary): local_runner = FrontendRunner() - # The agent instruction is set here instruction = f""" Answer user questions to the best of your knowledge using provided tools. @@ -118,105 +123,170 @@ def run_query_sync(request_text, cluster_name, location, instance_name, database parts=[types.Part.from_text(text=request_text)] ) - full_response = "" - grid_headers = [] - grid_rows = [] + response_text = "" + local_headers = [] + local_rows = [] + debug_logs = [] + query_history_list = [] + last_sql = "" + + current_session_id = f"session_mesop_{uuid.uuid4().hex[:8]}" + + # Python 3.7+ async run + try: + loop = asyncio.get_event_loop() + except RuntimeError: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + if loop.is_running(): + import nest_asyncio + nest_asyncio.apply() + + async_gen = local_runner.runner.run_async( + user_id="mesop_user", + session_id=session_id, + new_message=msg + ) - # We need to run the async runner in a sync context because Mesop event handlers are sync/generators - async def _run(): - response_text = "" - local_headers = [] - local_rows = [] - debug_logs = [] + while True: try: - async for event in local_runner.runner.run_async( - user_id="mesop_user", - session_id=session_id, - new_message=msg - ): - if event.content and event.content.parts: - for p in event.content.parts: - if p.text: - response_text += p.text + event = loop.run_until_complete(async_gen.__anext__()) + + if event.content and event.content.parts: + for p in event.content.parts: + if p.text: + response_text += p.text + + if p.function_call: + name = p.function_call.name + args = p.function_call.args + + # Resiliently extract sqlStatement + sql_statement = "" + if hasattr(args, "get"): + sql_statement = args.get("sqlStatement", "") + else: + try: + sql_statement = args.sqlStatement + except: + try: + sql_statement = args["sqlStatement"] + except: + pass + + if sql_statement: + import re + formatted_sql = sql_statement + keywords = ['FROM', 'WHERE', 'AND', 'OR', 'GROUP BY', 'ORDER BY', 'LIMIT', 'JOIN', 'LEFT JOIN', 'INNER JOIN'] + for kw in keywords: + formatted_sql = re.sub(rf'\b({kw})\b', r'\n\1', formatted_sql, flags=re.IGNORECASE) - # Intercept function responses to extract raw data - if hasattr(event, "get_function_responses"): - function_responses = event.get_function_responses() - for f_resp in function_responses: - # Depending on the ADK wrapper, it could be a raw dict or a structured object - data = [] - if hasattr(f_resp, "response") and hasattr(f_resp.response, "get"): - # typical genai format for part.function_response - res_dict = f_resp.response - elif isinstance(f_resp, dict): - res_dict = f_resp + last_sql = formatted_sql + log_entry = f"👉 **Function Call**: `{name}`\n```sql\n{formatted_sql}\n```" + else: + try: + args_str = json.dumps(args, indent=2) + except: + args_str = str(args) + log_entry = f"👉 **Function Call**: `{name}`\n```json\n{args_str}\n```" + debug_logs.insert(0, log_entry) + + if p.function_response: + name = p.function_response.name + response = p.function_response.response + + # Intercept function responses to extract raw data for table rendering + res_dict = {} + if isinstance(response, dict): + res_dict = response + elif hasattr(response, "get"): + res_dict = response else: try: - res_dict = f_resp.model_dump() + res_dict = response.model_dump() except: - res_dict = {} + pass - # Suppose AlloyDB MCP returns data under "results" or directly as list - # This is a heuristic based on standard tabular MCP returns rows = [] - if "results" in res_dict and isinstance(res_dict["results"], list): - rows = res_dict["results"] - elif isinstance(res_dict, list): - rows = res_dict - elif isinstance(res_dict.get("content"), list): + # 1. Try structuredContent first + if "structuredContent" in res_dict and isinstance(res_dict["structuredContent"], dict): + structured = res_dict["structuredContent"] + if "sqlResults" in structured and isinstance(structured["sqlResults"], list): + for result in structured["sqlResults"]: + if "rows" in result and "columns" in result: + headers = [c["name"] for c in result["columns"]] + for r in result["rows"]: + if "values" in r: + row_vals = [v.get("value", "") for v in r["values"]] + rows.append(dict(zip(headers, row_vals))) + + # 2. Fallback to text string parsing inside content + if not rows and isinstance(res_dict.get("content"), list): try: - content_list = res_dict["content"] - for c in content_list: + for c in res_dict["content"]: if isinstance(c, dict) and c.get("type") == "text": struct = json.loads(c.get("text", "{}")) - if isinstance(struct, list): - rows.extend(struct) - elif isinstance(struct, dict) and "results" in struct: - rows.extend(struct["results"]) + if "sqlResults" in struct and isinstance(struct["sqlResults"], list): + for result in struct["sqlResults"]: + if "rows" in result and "columns" in result: + headers = [c["name"] for c in result["columns"]] + for r in result["rows"]: + if "values" in r: + row_vals = [v.get("value", "") for v in r["values"]] + rows.append(dict(zip(headers, row_vals))) except: pass + # Always append query to the tabs, even if it has 0 rows or errored + local_headers = list(rows[0].keys()) if (rows and isinstance(rows[0], dict)) else ["Result"] + local_rows = [] + if rows and isinstance(rows[0], dict): - if not local_headers: - local_headers = list(rows[0].keys()) for row in rows: local_rows.append([str(row.get(h, "")) for h in local_headers]) + elif "structuredContent" in res_dict and "metadata" in res_dict["structuredContent"]: + msg = res_dict["structuredContent"]["metadata"].get("message", "Returned 0 rows") + local_rows.append([msg]) + else: + local_rows.append(["Returned 0 rows"]) + + query_history_list.append({ + "name": f"Query {len(query_history_list) + 1}", + "query": last_sql, + "headers": local_headers, + "rows": local_rows + }) - if hasattr(event, "get_function_calls"): - calls = event.get_function_calls() - if calls: - for c in calls: + # Compact Debug Log Formatting + if name == "execute_sql_read_only": + num_rows = len(rows) + preview = "" + if rows: + preview = json.dumps(rows[:3], indent=2) + if len(rows) > 3: + preview += f"\n... ({len(rows) - 3} more rows)" + log_entry = f"✅ **Function Response**: `execute_sql_read_only` - Returned {num_rows} row(s)\n```json\n{preview}\n```" + else: try: - if hasattr(c, "name"): - name = c.name - args = c.args - elif isinstance(c, dict): - name = c.get("name") - args = c.get("args") - else: - name = "Unknown" - args = str(c) - debug_logs.append(f"Function Call: {name}\nArguments: {json.dumps(args, indent=2)}") + resp_str = json.dumps(response, indent=2) except: - debug_logs.append(f"Function Call: {str(c)}") + resp_str = str(response) + if len(resp_str) > 1000: + resp_str = resp_str[:1000] + "\n... (truncated)" + log_entry = f"✅ **Function Response**: `{name}`\n```json\n{resp_str}\n```" + + debug_logs.insert(0, log_entry) + + yield response_text, local_headers, local_rows, "\n\n".join(debug_logs), query_history_list + except StopAsyncIteration: + break except Exception as e: response_text = f"Error: {str(e)}" - return response_text, local_headers, local_rows, "\n\n".join(debug_logs) - - # Python 3.7+ async run - try: - loop = asyncio.get_event_loop() - if loop.is_running(): - import nest_asyncio - nest_asyncio.apply() - full_response, grid_headers, grid_rows, debug_info = loop.run_until_complete(_run()) - else: - full_response, grid_headers, grid_rows, debug_info = asyncio.run(_run()) - except RuntimeError: - full_response, grid_headers, grid_rows, debug_info = asyncio.run(_run()) - - return full_response, grid_headers, grid_rows, debug_info + yield response_text, local_headers, local_rows, "\n\n".join(debug_logs), query_history_list + break + def submit_query(e: me.ClickEvent): state = me.state(State) @@ -227,9 +297,13 @@ def submit_query(e: me.ClickEvent): state.is_loading = True state.error_message = "" state.response_text = "" + state.debug_info = "" + state.grid_headers = [] + state.grid_rows = [] + state.query_history_json = "[]" + state.active_query_tab = 0 yield - # Since project_id isn't directly configured in the UI we can extract it from the agent module from data_agent.agent import project_id if not state.session_id: @@ -240,8 +314,9 @@ def submit_query(e: me.ClickEvent): if len(history_text) > 2000000: summary_prompt = f"Summarize the following conversation history concisely. Focus on the key facts and results retrieved so far:\n\n{history_text}" temp_session_id = f"session_summary_{uuid.uuid4().hex[:8]}" + summary_response = "" # Call without summary to avoid recursion - summary_response, _, _, _ = run_query_sync( + for resp_t, _, _, _, _ in run_query_sync( summary_prompt, state.cluster_name, state.location, @@ -250,13 +325,20 @@ def submit_query(e: me.ClickEvent): project_id, temp_session_id, "" - ) + ): + summary_response = resp_t + state.context_summary = summary_response state.conversation_history = [] # Generate a new session ID to effectively clear the ADK session history state.session_id = f"session_mesop_{uuid.uuid4().hex[:8]}" - response_text, headers, rows, debug_text = run_query_sync( + response_text = "" + headers = [] + rows = [] + debug_text = "" + + for resp_t, heads, rws, dbg_t, history in run_query_sync( state.request_text, state.cluster_name, state.location, @@ -265,14 +347,21 @@ def submit_query(e: me.ClickEvent): project_id, state.session_id, state.context_summary - ) - - state.response_text = response_text - + ): + response_text = resp_t + headers = heads + rows = rws + debug_text = dbg_t + + # Stream logs to UI so the user sees progress + state.debug_info = debug_text + state.response_text = response_text + state.query_history_json = json.dumps(history) + yield + # Append to history state.conversation_history.append(f"User: {state.request_text}") state.conversation_history.append(f"Agent: {response_text}") - # Fallback to parse JSON from the final response text if no tools intercepted it if not headers and not rows and "{" in response_text and "}" in response_text: start = response_text.find('{') @@ -314,7 +403,6 @@ def submit_query(e: me.ClickEvent): state.grid_headers = headers state.grid_rows = rows - state.debug_info = debug_text state.is_loading = False request_lower = state.request_text.lower() @@ -333,7 +421,7 @@ def submit_query(e: me.ClickEvent): @me.page( path="/", on_load=on_load, - title="Cymbal Logistic Agent", + title="Cymbal Logistics Agent", security_policy=sec_policy, ) def app(): @@ -351,20 +439,33 @@ def app(): link.type = 'image/png'; link.href = '/static/cymbal_logo_v2.png?v=2'; document.head.appendChild(link); - document.title = "Cymbal Logistic Agent"; + document.title = "Cymbal Logistics Agent"; }, 100); """) @@ -372,21 +473,22 @@ def app(): display="flex", flex_direction="row", height="100vh", - background="#F0F8FF", # light alice blue - font_family="Google Sans, Roboto, sans-serif", - color="#202124" + background="linear-gradient(135deg, #F5F4F0 0%, #EAE7E0 100%)", + font_family="'Outfit', 'Inter', 'Google Sans', sans-serif", + color="#111111" )): # Left Sidebar (Configurations) with me.box(style=me.Style( - width="300px", - background="#E3F2FD", # light blue + width="320px", + background="rgba(255, 255, 255, 0.4)", + backdrop_filter="blur(20px)", padding=me.Padding.all(24), - border=me.Border(right=me.BorderSide(width=1, style="solid", color="#BBDEFB")), + border=me.Border(right=me.BorderSide(width=1, style="solid", color="rgba(255, 255, 255, 0.3)")), display="flex", flex_direction="column", gap=16 )): - me.text("Database Config", type="headline-6", style=me.Style(margin=me.Margin(bottom=16), color="#1A73E8", font_weight="500")) + me.text("Database Config", type="headline-6", style=me.Style(margin=me.Margin(bottom=16), color="#000000", font_weight="600")) me.input( label="Cluster Name", @@ -426,26 +528,27 @@ def app(): )): # Hero Header with me.box(style=me.Style( - width="100%", - height="120px", - margin=me.Margin(top=16), - border_radius=8, - background="linear-gradient(to bottom, #64B5F6 0%, #1976D2 100%)", + height="100px", + margin=me.Margin(top=24, left=32, right=32), + border_radius=16, + background="rgba(255, 255, 255, 0.6)", + backdrop_filter="blur(20px)", display="flex", flex_direction="row", align_items="center", - padding=me.Padding(left=32, right=32), - box_shadow="0 2px 4px rgba(0,0,0,0.1)" + padding=me.Padding.symmetric(horizontal=24), + border=me.Border.all(me.BorderSide(width=1, style="solid", color="rgba(255, 255, 255, 0.5)")), + box_shadow="0 8px 32px rgba(0, 0, 0, 0.04)" )): me.image( src="/static/cymbal_logo_v2.png", style=me.Style( - height="80px", - margin=me.Margin(right=24), + height="60px", + margin=me.Margin(right=16), border_radius=8 ) ) - me.text("Cymbal Logistic Agent", type="headline-3", style=me.Style(color="#FFFFFF", font_weight="600", margin=me.Margin(top=0, bottom=0, left=0, right=0))) + me.text("Cymbal Logistics Agent", style=me.Style(color="#000000", font_weight=900, font_family="'Avant Garde', 'Avantgarde', 'Century Gothic', sans-serif", font_size="36px", letter_spacing="-1.5px", margin=me.Margin(top=0, bottom=0, left=0, right=0))) # Content Area with me.box(style=me.Style( @@ -456,11 +559,12 @@ def app(): )): # Top Right: Input Area with me.box(style=me.Style( - background="#E3F2FD", # light blue + background="rgba(255, 255, 255, 0.7)", + backdrop_filter="blur(20px)", padding=me.Padding.all(24), - border_radius=8, - border=me.Border.all(me.BorderSide(width=1, style="solid", color="#BBDEFB")), - box_shadow="0 1px 2px 0 rgba(60,64,67,0.3), 0 1px 3px 1px rgba(60,64,67,0.15)", + border_radius=16, + border=me.Border.all(me.BorderSide(width=1, style="solid", color="rgba(255, 255, 255, 0.6)")), + box_shadow="0 8px 32px rgba(0, 0, 0, 0.04)", display="flex", flex_direction="column", gap=16 @@ -481,52 +585,92 @@ def app(): style=me.Style(width="100%", border_radius=4) ) - with me.box(style=me.Style(display="flex", justify_content="flex-end")): - me.button( - "Submit Request", + with me.box(style=me.Style(display="flex", justify_content="space-between", align_items="center")): + # Left Side: Spinner (if loading) + if state.is_loading: + with me.box(style=me.Style(display="flex", flex_direction="row", gap=12, align_items="center")): + me.progress_spinner() + me.text("Agent is thinking...", style=me.Style(color="#5F6368", font_weight="500")) + else: + me.box() # Empty box to keep space-between working + + with me.box( + classes="custom-btn", on_click=submit_query, - type="raised", - color="primary" - ) + style=me.Style( + background="#000000", + color="#FFFFFF", + padding=me.Padding.symmetric(vertical=12, horizontal=32), + border_radius=8, + cursor="pointer", + display="flex", + align_items="center", + justify_content="center" + ) + ): + me.text("Submit Request", style=me.Style(font_weight="500", font_size="14px")) if state.error_message: me.text(state.error_message, style=me.Style(color="#D93025", font_weight="500")) # Bottom Right: Results Area - if state.is_loading: - with me.box(style=me.Style(display="flex", flex_direction="row", gap=12, align_items="center")): - me.progress_spinner() - me.text("Agent is thinking...", style=me.Style(color="#5F6368", font_weight="500")) - - elif state.response_text: + if state.response_text or state.debug_info: with me.box(style=me.Style( - background="#E3F2FD", # light blue + background="rgba(255, 255, 255, 0.7)", + backdrop_filter="blur(20px)", padding=me.Padding.all(24), - border_radius=8, - border=me.Border.all(me.BorderSide(width=1, style="solid", color="#BBDEFB")), - box_shadow="0 1px 2px 0 rgba(60,64,67,0.3), 0 1px 3px 1px rgba(60,64,67,0.15)", + border_radius=16, + border=me.Border.all(me.BorderSide(width=1, style="solid", color="rgba(255, 255, 255, 0.6)")), + box_shadow="0 8px 32px rgba(0, 0, 0, 0.04)", display="flex", flex_direction="column", gap=24 )): - if state.has_chart and state.grid_headers and len(state.grid_headers) >= 2: - me.text("Data Visualization", type="subtitle-1", style=me.Style(font_weight="bold", color="#1A73E8")) - else: - me.text("Response", type="subtitle-1", style=me.Style(font_weight="bold", color="#1A73E8")) - me.markdown(state.response_text) + if state.response_text: + if state.has_chart and state.grid_headers and len(state.grid_headers) >= 2: + me.text("Data Visualization", type="subtitle-1", style=me.Style(font_weight="bold", color="#000000")) + else: + me.text("Response", type="subtitle-1", style=me.Style(font_weight="bold", color="#000000")) + me.markdown(state.response_text) - if state.grid_headers: - me.text("Data Query Results", type="subtitle-1", style=me.Style(font_weight="bold", margin=me.Margin(top=16))) - with me.box(style=me.Style(max_height="400px", overflow_y="auto")): - # me.table format accepts Pandas DataFrames directly - try: - df_table = pd.DataFrame(state.grid_rows, columns=state.grid_headers) - me.table(df_table, header=me.TableHeader(sticky=True)) - except Exception as e: - me.text(f"Could not render table: {str(e)}", style=me.Style(color="red")) + query_history = json.loads(state.query_history_json) if state.query_history_json else [] + if query_history and not state.is_loading: + me.text("Data Query Results", type="subtitle-1", style=me.Style(font_weight="bold", margin=me.Margin(top=16), color="#000000")) + + # Tab Strip + with me.box(style=me.Style(display="flex", flex_direction="row", gap=8, margin=me.Margin(bottom=16), overflow_x="auto")): + for idx, q in enumerate(query_history): + is_active = (idx == state.active_query_tab) + with me.box( + key=str(idx), + on_click=select_tab, + style=me.Style( + padding=me.Padding.symmetric(vertical=8, horizontal=16), + border_radius=8, + background="#000000" if is_active else "rgba(0, 0, 0, 0.05)", + color="#FFFFFF" if is_active else "#000000", + cursor="pointer", + font_weight="500", + font_size="13px", + white_space="nowrap" + ) + ): + me.text(q["name"]) + + # Active Tab Content + active_tab = query_history[state.active_query_tab] if state.active_query_tab < len(query_history) else None + if active_tab: + if "query" in active_tab and active_tab["query"]: + me.markdown(f"```sql\n{active_tab['query']}\n```") + with me.box(style=me.Style(max_height="400px", overflow_y="auto", margin=me.Margin(top=12))): + try: + df_table = pd.DataFrame(active_tab["rows"], columns=active_tab["headers"]) + me.table(df_table, header=me.TableHeader(sticky=True)) + except Exception as e: + me.text(f"Could not render table: {str(e)}", style=me.Style(color="red")) if state.has_chart and state.grid_headers and len(state.grid_headers) >= 2: - me.text("Chart Analysis", type="subtitle-1", style=me.Style(font_weight="bold", margin=me.Margin(top=16))) + me.text("Chart Analysis", type="subtitle-1", style=me.Style(font_weight="bold", margin=me.Margin(top=16), color="#000000")) try: # Plot a simple bar chart using the first categorical col and first numeric col @@ -562,7 +706,7 @@ def app(): # Sort by numeric column d = d.sort_values(by=numeric_col, ascending=False).head(15) - ax.bar(d[category_col].astype(str), d[numeric_col], color="#1976D2") + ax.bar(d[category_col].astype(str), d[numeric_col], color="#000000") ax.set_ylabel(numeric_col) ax.set_xlabel(category_col) ax.set_title(f"Chart: {numeric_col} by {category_col}") @@ -578,14 +722,12 @@ def app(): is_debug_env = os.environ.get("DEBUG", "false").lower() == "true" is_debug = is_debug_env or state.enable_debug if is_debug and state.debug_info: - me.text("Debug Execution Logs", type="subtitle-1", style=me.Style(font_weight="bold", margin=me.Margin(top=16))) + me.text("Debug Execution Logs", type="subtitle-1", style=me.Style(font_weight="bold", margin=me.Margin(top=16), color="#000000")) with me.box(style=me.Style( - background="#333333", - color="#00FF00", + background="rgba(0, 0, 0, 0.02)", padding=me.Padding.all(16), - border_radius=8, - font_family="monospace", - font_size="14px", - white_space="pre-wrap" + border_radius=12, + border=me.Border.all(me.BorderSide(width=1, style="solid", color="rgba(0, 0, 0, 0.05)")), + font_size="14px" )): - me.text(state.debug_info) + me.markdown(state.debug_info) diff --git a/alloydb-ai-mcp/cymbal_logistic/static/cymbal_logo_v2.png b/alloydb-ai-mcp/cymbal_logistic/static/cymbal_logo_v2.png index fd655fd..325decb 100644 Binary files a/alloydb-ai-mcp/cymbal_logistic/static/cymbal_logo_v2.png and b/alloydb-ai-mcp/cymbal_logistic/static/cymbal_logo_v2.png differ