import json from agent_build_sdk.server.server import EndpointServer from chinatown.chinatown_agent import ChinatownLLMAgent from fastapi.responses import HTMLResponse import html as html_lib def register_debug_endpoints(server: EndpointServer): agent = server.agent @server.app.get("/debug/requests") def get_requests(): history = agent.get_req_resp_history() return { "success": True, "total": len(history), "history": history, } @server.app.get("/debug/requests/view") def view_requests(): history = agent.get_req_resp_history() html_items = "" for i, item in enumerate(reversed(history)): req_data = item.get("req", {}) resp_data = item.get("resp", {}) req_type = item.get("type", "?") status = req_data.get("status", "?") round_num = req_data.get("round", "?") action = resp_data.get("action", "-") if resp_data else "-" html_items += f"""
{req_type} {html_lib.escape(str(status))} | Round {html_lib.escape(str(round_num))} | Action: {html_lib.escape(str(action))}
Request
{html_lib.escape(str(req_data))}
Response
{html_lib.escape(str(resp_data))}
""" return HTMLResponse(content=f""" Request/Response History

Request/Response History ({len(history)} items)

{html_items} """) @server.app.post("/debug/requests/clear") def clear_requests(): agent.clear_req_resp_history() return {"success": True, "message": "请求历史已清空"} if hasattr(agent, 'get_prompt_history') and callable(getattr(agent, 'get_prompt_history')): @server.app.get("/debug/prompts") def get_prompts(): history = agent.get_prompt_history() return { "success": True, "total": len(history), "prompts": history, } @server.app.get("/debug/prompts/view") def view_prompts(): history = agent.get_prompt_history() def format_content(data) -> str: """将请求体/响应体格式化为带换行标记的 HTML""" if isinstance(data, (dict, list)): text = json.dumps(data, ensure_ascii=False, indent=2) else: text = str(data) escaped = html_lib.escape(text) return escaped.replace("\\n", '\\n\n') html_items = "" for i, item in enumerate(reversed(history)): request_data = item.get("request") response_data = item.get("response", "") html_items += f"""
{html_lib.escape(str(item.get('elapsed_sec', 0)))}s | {html_lib.escape(str(item.get('timestamp', '')))}
Request
{format_content(request_data)}
Response
{format_content(response_data)}
""" return HTMLResponse(content=f""" LLM Call History

LLM Call History ({len(history)} items)

{html_items} """) @server.app.post("/debug/prompts/clear") def clear_prompts(): agent.clear_prompt_history() return {"success": True, "message": "提示词历史已清空"} if __name__ == '__main__': name = 'chinatown' agent = ChinatownLLMAgent(player_id="") server = EndpointServer(agent) register_debug_endpoints(server) server.start()