from typing import Any, Dict, List import gradio as gr from src.agent import run_agent, run_agent_iter from src.audit import read_audit_records from src.demo_presentation import ( APP_CSS, ARCHITECTURE_MARKDOWN, AUTHORITY_HTML, BUSINESS_CASE_HTML, HERO_HTML, SECURITY_MARKDOWN, TOOL_BELT_HTML, idle_activity_html, render_activity, render_summary, starting_activity_html, ) from src.schemas import AgentRunEvent, AgentRunResult def get_run_audit_records( call_ids: List[str], ) -> List[Dict[str, Any]]: """Return audit records belonging only to the current agent run.""" if not call_ids: return [] records = read_audit_records() matching_records = [ record for record in records if record.call.call_id in call_ids ] return [ record.model_dump(mode="json") for record in matching_records ] def handle_request(user_request: str): """Preserve the original programmatic callback contract.""" if not user_request or not user_request.strip(): return ( "Please enter a request.", [], [], [], ) try: result = run_agent(user_request.strip()) tool_calls = [ call.model_dump(mode="json") for call in result.tool_calls ] tool_results = [ tool_result.model_dump(mode="json") for tool_result in result.tool_results ] call_ids = [ call.call_id for call in result.tool_calls ] return ( result.final_answer, tool_calls, tool_results, get_run_audit_records(call_ids), ) except Exception: return ( ( "Agent execution failed. " "Please review the server logs for details." ), [], [], [], ) def _current_calls(events: list[AgentRunEvent]) -> list[Dict[str, Any]]: calls: dict[str, Dict[str, Any]] = {} for event in events: if event.call is not None: calls[event.call.call_id] = event.call.model_dump(mode="json") return list(calls.values()) def _current_results(events: list[AgentRunEvent]) -> list[Dict[str, Any]]: results: dict[str, Dict[str, Any]] = {} for event in events: if event.result is not None: results[event.result.call_id] = event.result.model_dump(mode="json") return list(results.values()) def _stream_outputs( events: list[AgentRunEvent], final_result: AgentRunResult | None = None, complete: bool = False, ): calls = ( [call.model_dump(mode="json") for call in final_result.tool_calls] if final_result is not None else _current_calls(events) ) results = ( [result.model_dump(mode="json") for result in final_result.tool_results] if final_result is not None else _current_results(events) ) call_ids = [call["call_id"] for call in calls] audits = get_run_audit_records(call_ids) if complete else [] answer = ( final_result.final_answer if final_result is not None else "*The final business answer will appear after controlled execution completes.*" ) return ( render_activity(events, complete=complete), answer, render_summary(final_result), calls, results, audits, ) def stream_request(user_request: str): """Stream real model/application boundary events into the Gradio demo.""" if not user_request or not user_request.strip(): yield ( idle_activity_html(), "Please enter a request.", render_summary(None), [], [], [], ) return events: list[AgentRunEvent] = [] yield ( starting_activity_html(), "*The final business answer will appear after controlled execution completes.*", render_summary(None), [], [], [], ) try: for event in run_agent_iter(user_request.strip()): events.append(event) final_result = event.final_result complete = final_result is not None yield _stream_outputs( events=events, final_result=final_result, complete=complete, ) except Exception: error_event = AgentRunEvent( event="max_rounds_reached", message=( "The request could not complete. The application stopped the run " "without exposing internal server details." ), ) events.append(error_event) yield ( render_activity(events, complete=True), ( "Agent execution failed. Please review the server logs for details." ), render_summary(None), _current_calls(events), _current_results(events), [], ) FLAGSHIP_REQUEST = ( "We may ship equipment to Japan. Find the Electronics items currently in " "inventory, calculate an accessory budget of $347 per matching item, and " "give me Japan's capital, region, and income classification." ) with gr.Blocks( title="Governed Tool-Using Agent", css=APP_CSS, ) as demo: with gr.Column(elem_classes=["agent-shell"]): gr.HTML(HERO_HTML) gr.HTML(BUSINESS_CASE_HTML) gr.Markdown("## The approved tool belt") gr.HTML(TOOL_BELT_HTML) gr.Markdown("## Who controls what?") gr.HTML(AUTHORITY_HTML) gr.Markdown("## Try a controlled business request") request_box = gr.Textbox( label="Operations request", value=FLAGSHIP_REQUEST, placeholder="Ask for work that may require one or more approved tools.", lines=4, ) gr.Examples( examples=[ [FLAGSHIP_REQUEST], ["What is 347 multiplied by 29?"], ["What electronics are currently in inventory?"], ["What is the capital, region, and income level of Japan?"], [ ( "How many Electronics items are in inventory, " "and what is 347 multiplied by that number?" ) ], ], inputs=request_box, ) run_button = gr.Button( "Run controlled request", variant="primary", ) activity_output = gr.HTML(idle_activity_html()) with gr.Tabs(): with gr.Tab("Business Result"): gr.Markdown( "The answer below is produced only after requested capabilities " "pass through the application-controlled boundary." ) final_answer = gr.Markdown( "*Run a request to produce a business answer.*" ) summary_output = gr.HTML(render_summary(None)) with gr.Tab("Engineering Audit"): gr.Markdown( "These are the structured requests, normalized results, and " "current-run audit records behind the business-facing view." ) tool_calls_output = gr.JSON(label="Model-proposed Tool Calls") tool_results_output = gr.JSON(label="Controlled Tool Results") audit_output = gr.JSON(label="Current Run Audit Records") with gr.Tab("Security & Failure Semantics"): gr.Markdown(SECURITY_MARKDOWN) with gr.Tab("Architecture"): gr.Markdown(ARCHITECTURE_MARKDOWN) run_button.click( fn=stream_request, inputs=request_box, outputs=[ activity_output, final_answer, summary_output, tool_calls_output, tool_results_output, audit_output, ], show_progress="hidden", ) if __name__ == "__main__": demo.launch()