""" Second Opinion AI Agent - HuggingFace Spaces Version Agentic framework that uses MCP Server tools for deep analysis Supports multiple LLM providers: Anthropic, OpenAI, and Google Gemini """ import gradio as gr import os from typing import List, Dict, Tuple, Optional, Callable, Any import json from datetime import datetime import re # Import MCP server to access tools directly from mcp_server import ( analyze_assumptions, detect_cognitive_biases, generate_alternatives, perform_premortem_analysis, identify_stakeholders_and_impacts, second_order_thinking, opportunity_cost_analysis, red_team_analysis ) # ============================================================================= # MCP TOOLS REGISTRY - Direct access to MCP server tools # ============================================================================= MCP_TOOLS = { "analyze_assumptions": { "function": analyze_assumptions, "description": "Analyzes an idea to identify hidden assumptions and unstated premises. Use this when you need to uncover what's being taken for granted.", "parameters": { "idea": "The idea or decision to analyze (required)", "context": "Additional context or background information (optional)" } }, "detect_cognitive_biases": { "function": detect_cognitive_biases, "description": "Detects potential cognitive biases in reasoning and decision-making. Use this to identify confirmation bias, anchoring, sunk cost fallacy, etc.", "parameters": { "idea": "The idea or decision being proposed (required)", "reasoning": "The reasoning or justification provided (optional)" } }, "generate_alternatives": { "function": generate_alternatives, "description": "Generates alternative approaches and solutions to consider. Use this to explore other options beyond the proposed idea.", "parameters": { "idea": "The original idea or approach (required)", "constraints": "Known constraints or requirements (optional)", "num_alternatives": "Number of alternatives to generate, 1-10 (optional, default 5)" } }, "perform_premortem_analysis": { "function": perform_premortem_analysis, "description": "Performs a pre-mortem analysis: imagine the idea failed and identify why. Use this to anticipate failure modes.", "parameters": { "idea": "The idea or project to analyze (required)", "timeframe": "When to imagine the failure, e.g. '6 months', '1 year' (optional)" } }, "identify_stakeholders_and_impacts": { "function": identify_stakeholders_and_impacts, "description": "Identifies all stakeholders and analyzes potential impacts on each group. Use this to understand who is affected.", "parameters": { "idea": "The idea or decision to analyze (required)", "organization_context": "Context about the organization or situation (optional)" } }, "second_order_thinking": { "function": second_order_thinking, "description": "Analyzes second and third-order consequences of an idea. Use this to explore cascading effects and long-term implications.", "parameters": { "idea": "The idea or decision to analyze (required)", "time_horizon": "Time period to consider, e.g. '2-5 years' (optional)" } }, "opportunity_cost_analysis": { "function": opportunity_cost_analysis, "description": "Analyzes opportunity costs: what you give up by choosing this path. Use this to understand trade-offs.", "parameters": { "idea": "The idea or decision being considered (required)", "resources": "Available resources like time, money, people (optional)", "alternatives": "Other options being considered (optional)" } }, "red_team_analysis": { "function": red_team_analysis, "description": "Performs red team analysis: actively tries to break or exploit the idea. Use this to find vulnerabilities.", "parameters": { "idea": "The idea, system, or plan to attack (required)", "attack_surface": "Known vulnerabilities or areas of concern (optional)" } } } def get_tools_description() -> str: """Generate a formatted description of all available tools for the agent""" tools_desc = [] for name, tool in MCP_TOOLS.items(): params = ", ".join([f"{k}: {v}" for k, v in tool["parameters"].items()]) tools_desc.append(f"- {name}({params})\n Description: {tool['description']}") return "\n\n".join(tools_desc) def call_mcp_tool(tool_name: str, **kwargs) -> str: """Call an MCP tool by name with given arguments""" if tool_name not in MCP_TOOLS: return json.dumps({"error": f"Unknown tool: {tool_name}"}) try: tool_func = MCP_TOOLS[tool_name]["function"] result = tool_func(**kwargs) return result except Exception as e: return json.dumps({"error": f"Tool execution failed: {str(e)}"}) # ============================================================================= # AGENT SYSTEM PROMPTS # ============================================================================= AGENT_SYSTEM_PROMPT = """You are a Second Opinion AI Agent - an expert critical thinking assistant that helps people challenge their ideas and decisions. You have access to the following analysis tools from the MCP server: {tools} ## HOW TO USE TOOLS When you need to use a tool, respond with a tool call in this EXACT format: {{"tool": "tool_name", "arguments": {{"param1": "value1", "param2": "value2"}}}} You can call multiple tools in sequence. After each tool call, you'll receive the results, and you can then call another tool or provide your final analysis. ## YOUR APPROACH 1. **Analyze the Request**: First understand what the user is asking for help with 2. **Select Appropriate Tools**: Choose 2-4 tools that would provide the most valuable analysis 3. **Call Tools**: Execute tool calls one at a time 4. **Synthesize Insights**: After gathering tool outputs, synthesize them into a cohesive, actionable analysis ## RESPONSE GUIDELINES - Start by briefly acknowledging the user's idea/decision - Use tools to gather structured analysis - After tool results, provide a SYNTHESIZED analysis that: - Highlights the most critical insights - Identifies key blind spots or risks - Offers specific, actionable recommendations - Maintains a constructive but challenging tone Remember: Your goal is to help the user see what they might be missing, not to simply validate or criticize their thinking. When you have gathered enough insights and are ready to provide your final analysis, respond WITHOUT any tool calls.""" PERSONA_INSTRUCTIONS = { "Devil's Advocate": """ ## PERSONA: DEVIL'S ADVOCATE Your role is to challenge every assumption and argue the opposite perspective. Focus on: Counter-arguments, hidden flaws, risks, and alternative viewpoints. Recommended tools: detect_cognitive_biases, analyze_assumptions, red_team_analysis""", "Strategic Thinker": """ ## PERSONA: STRATEGIC THINKER Your role is to analyze long-term consequences and strategic implications. Focus on: Second-order effects, opportunity costs, timing, and strategic gaps. Recommended tools: second_order_thinking, opportunity_cost_analysis, perform_premortem_analysis""", "Pragmatic Realist": """ ## PERSONA: PRAGMATIC REALIST Your role is to ground ideas in practical reality and implementation challenges. Focus on: Feasibility, resource requirements, execution gaps, and real-world constraints. Recommended tools: perform_premortem_analysis, identify_stakeholders_and_impacts, generate_alternatives""", "Systems Thinker": """ ## PERSONA: SYSTEMS THINKER Your role is to examine interconnections and systemic effects. Focus on: Feedback loops, unintended consequences, stakeholder impacts, and leverage points. Recommended tools: second_order_thinking, identify_stakeholders_and_impacts, red_team_analysis""", "Comprehensive Analyst": """ ## PERSONA: COMPREHENSIVE ANALYST Your role is to provide thorough multi-dimensional analysis. Focus on: All aspects - assumptions, biases, consequences, stakeholders, and alternatives. Recommended tools: Use ALL available tools for comprehensive coverage""" } # ============================================================================= # LLM PROVIDER CONFIGURATIONS # ============================================================================= LLM_PROVIDERS = { "Anthropic (Claude)": { "models": ["claude-sonnet-4-5-20250929", "claude-haiku-4-5-20251001"], "default_model": "claude-sonnet-4-5-20250929", "env_key": "ANTHROPIC_API_KEY" }, "OpenAI (GPT)": { "models": ["gpt-4.1", "gpt-4o", "gpt-4.1-mini"], "default_model": "gpt-4o", "env_key": "OPENAI_API_KEY" }, "Google (Gemini)": { "models": ["gemini-2.5-flash", "gemini-2.5-pro", "gemini-2.5-flash-lite"], "default_model": "gemini-2.5-flash", "env_key": "GOOGLE_API_KEY" } } def get_client(provider: str, api_key: str): """Initialize and return the appropriate client based on provider""" if not api_key: return None, "API key is required" try: if provider == "Anthropic (Claude)": import anthropic return anthropic.Anthropic(api_key=api_key), None elif provider == "OpenAI (GPT)": from openai import OpenAI return OpenAI(api_key=api_key), None elif provider == "Google (Gemini)": import google.generativeai as genai genai.configure(api_key=api_key) return genai, None else: return None, f"Unknown provider: {provider}" except Exception as e: return None, f"Failed to initialize client: {str(e)}" # ============================================================================= # LLM CALL FUNCTIONS # ============================================================================= def call_anthropic(client, model: str, system_prompt: str, messages: List[Dict], temperature: float, max_tokens: int) -> str: """Call Anthropic API""" response = client.messages.create( model=model, max_tokens=max_tokens, temperature=temperature, system=system_prompt, messages=messages ) return response.content[0].text def call_openai(client, model: str, system_prompt: str, messages: List[Dict], temperature: float, max_tokens: int) -> str: """Call OpenAI API""" openai_messages = [{"role": "system", "content": system_prompt}] for msg in messages: openai_messages.append({ "role": msg["role"], "content": msg["content"] }) response = client.chat.completions.create( model=model, messages=openai_messages, temperature=temperature, max_tokens=max_tokens ) return response.choices[0].message.content def call_gemini(genai, model: str, system_prompt: str, messages: List[Dict], temperature: float, max_tokens: int) -> str: """Call Google Gemini API""" generation_config = { "temperature": temperature, "max_output_tokens": max_tokens, } model_instance = genai.GenerativeModel( model_name=model, system_instruction=system_prompt, generation_config=generation_config ) contents = [] for msg in messages: role = "user" if msg["role"] == "user" else "model" contents.append({ "role": role, "parts": [{"text": msg["content"]}] }) response = model_instance.generate_content(contents) return response.text def call_llm(client, provider: str, model: str, system_prompt: str, messages: List[Dict], temperature: float, max_tokens: int) -> str: """Universal LLM call function""" if provider == "Anthropic (Claude)": return call_anthropic(client, model, system_prompt, messages, temperature, max_tokens) elif provider == "OpenAI (GPT)": return call_openai(client, model, system_prompt, messages, temperature, max_tokens) elif provider == "Google (Gemini)": return call_gemini(client, model, system_prompt, messages, temperature, max_tokens) else: raise ValueError(f"Unknown provider: {provider}") # ============================================================================= # AGENT EXECUTION ENGINE # ============================================================================= def parse_tool_calls(response: str) -> List[Dict]: """Extract tool calls from agent response""" tool_calls = [] pattern = r'\s*(\{.*?\})\s*' matches = re.findall(pattern, response, re.DOTALL) for match in matches: try: tool_call = json.loads(match) if "tool" in tool_call: tool_calls.append(tool_call) except json.JSONDecodeError: continue return tool_calls def format_tool_result(tool_name: str, result: str) -> str: """Format tool result for inclusion in conversation""" try: parsed = json.loads(result) formatted = json.dumps(parsed, indent=2) except: formatted = result return f""" {formatted} """ def run_agent( user_input: str, persona: str, client, provider: str, model: str, conversation_history: List[Dict], temperature: float = 0.7, max_tokens: int = 4000, max_iterations: int = 6 ) -> Tuple[str, List[Dict], List[str]]: """ Run the agentic loop with MCP tool calling Returns: Tuple of (final_response, updated_conversation_history, tool_execution_log) """ # Build system prompt with tools and persona tools_desc = get_tools_description() system_prompt = AGENT_SYSTEM_PROMPT.format(tools=tools_desc) system_prompt += "\n\n" + PERSONA_INSTRUCTIONS.get(persona, PERSONA_INSTRUCTIONS["Comprehensive Analyst"]) # Start with conversation history and add new user message agent_messages = list(conversation_history) agent_messages.append({"role": "user", "content": user_input}) tool_execution_log = [] iteration = 0 final_response = "" while iteration < max_iterations: iteration += 1 # Call LLM try: response = call_llm( client, provider, model, system_prompt, agent_messages, temperature, max_tokens ) except Exception as e: return f"⚠️ Error calling {provider} API: {str(e)}", conversation_history, tool_execution_log # Check for tool calls tool_calls = parse_tool_calls(response) if not tool_calls: # No tool calls - this is the final response final_response = response break # Execute tool calls tool_results = [] for tc in tool_calls: tool_name = tc.get("tool", "") arguments = tc.get("arguments", {}) tool_execution_log.append(f"🔧 Calling: {tool_name}") # Execute the MCP tool result = call_mcp_tool(tool_name, **arguments) tool_results.append(format_tool_result(tool_name, result)) tool_execution_log.append(f"✓ {tool_name} completed") # Add assistant response and tool results to messages agent_messages.append({"role": "assistant", "content": response}) agent_messages.append({ "role": "user", "content": "Tool execution results:\n\n" + "\n\n".join(tool_results) + "\n\nPlease continue your analysis. Call more tools if needed, or provide your final synthesized analysis." }) if not final_response: final_response = "⚠️ Agent reached maximum iterations without completing analysis." # Clean the final response - remove any remaining tool call tags final_response = re.sub(r'.*?', '', final_response, flags=re.DOTALL).strip() # Update conversation history with user input and final response only updated_history = conversation_history + [ {"role": "user", "content": user_input}, {"role": "assistant", "content": final_response} ] return final_response, updated_history, tool_execution_log # ============================================================================= # MAIN INTERFACE FUNCTION # ============================================================================= def normalize_history(history) -> List[Dict]: """ Convert history to messages format regardless of input format. Handles: None, empty list, tuples format, messages format """ if history is None or history == []: return [] # Check if it's already in messages format if isinstance(history, list) and len(history) > 0: first_item = history[0] # Messages format: [{"role": "...", "content": "..."}, ...] if isinstance(first_item, dict) and "role" in first_item: return [msg for msg in history if isinstance(msg, dict) and "role" in msg and "content" in msg] # Tuple format: [("user msg", "assistant msg"), ...] if isinstance(first_item, (tuple, list)) and len(first_item) == 2: messages = [] for item in history: if isinstance(item, (tuple, list)) and len(item) == 2: user_msg, assistant_msg = item if user_msg: messages.append({"role": "user", "content": str(user_msg)}) if assistant_msg: messages.append({"role": "assistant", "content": str(assistant_msg)}) return messages return [] def format_history_for_gradio(messages: List[Dict]) -> List[Dict]: """ Format messages for Gradio Chatbot (messages format). Ensures all messages have proper structure. """ result = [] for msg in messages: if isinstance(msg, dict) and "role" in msg and "content" in msg: result.append({ "role": str(msg["role"]), "content": str(msg["content"]) if msg["content"] else "" }) return result def get_second_opinion( user_input: str, persona: str, provider: str, model: str, api_key: str, chatbot_history, # Can be any format temperature: float = 0.7, max_tokens: int = 4000 ) -> Tuple[str, List[Dict], str]: """ Get a second opinion from the AI agent using MCP tools Returns: Tuple of (response, updated_history, tool_log_display) """ # Normalize history to messages format history = normalize_history(chatbot_history) if not api_key: env_key = LLM_PROVIDERS.get(provider, {}).get("env_key", "") api_key = os.environ.get(env_key, "") if not api_key: error_msg = f"⚠️ API key required. Please enter your {provider} API key or set {env_key} in HuggingFace Spaces Settings." return error_msg, format_history_for_gradio(history), "" client, error = get_client(provider, api_key) if error: return f"⚠️ Error: {error}", format_history_for_gradio(history), "" # Run the agent response, updated_history, tool_log = run_agent( user_input=user_input, persona=persona, client=client, provider=provider, model=model, conversation_history=history, temperature=temperature, max_tokens=max_tokens ) # Format tool log for display tool_log_display = "\n".join(tool_log) if tool_log else "No tools called" return response, format_history_for_gradio(updated_history), tool_log_display # ============================================================================= # GRADIO INTERFACE # ============================================================================= def create_interface(): """Create the Gradio interface""" custom_css = """ """ # Persona descriptions for UI PERSONA_DESCRIPTIONS = { "Devil's Advocate": "Challenges assumptions and argues the opposite perspective", "Strategic Thinker": "Focuses on long-term consequences and strategic implications", "Pragmatic Realist": "Grounds ideas in practical reality and implementation challenges", "Systems Thinker": "Examines interconnections and systemic effects", "Comprehensive Analyst": "Provides thorough multi-dimensional analysis" } with gr.Blocks(title="The Second Opinion") as app: gr.HTML(custom_css) gr.Markdown( """ # 🧠 The Second Opinion

Agentic analysis powered by MCP tools • Challenge your thinking • Catch blind spots • Discover alternatives

""" ) with gr.Row(): with gr.Column(scale=1): # Provider Section with gr.Group(elem_classes="main-section"): gr.Markdown("### 🤖 Provider") provider_dropdown = gr.Dropdown( choices=list(LLM_PROVIDERS.keys()), value="Google (Gemini)", label="AI Provider", ) model_dropdown = gr.Dropdown( choices=LLM_PROVIDERS["Google (Gemini)"]["models"], value=LLM_PROVIDERS["Google (Gemini)"]["default_model"], label="Model", ) api_key_input = gr.Textbox( label="API Key", placeholder="Enter your API key...", type="password", info="Your key is never stored" ) # Challenger Section with gr.Group(elem_classes="main-section"): gr.Markdown("### 🎭 Challenger") persona_dropdown = gr.Dropdown( choices=list(PERSONA_DESCRIPTIONS.keys()), value="Comprehensive Analyst", label="Persona", info="Select your critical thinking style" ) persona_description = gr.Markdown( value="
💡 " + PERSONA_DESCRIPTIONS["Comprehensive Analyst"] + "
" ) # Settings Section with gr.Group(elem_classes="main-section"): gr.Markdown("### ⚙️ Settings") temperature_slider = gr.Slider( minimum=0.0, maximum=1.0, value=0.5, step=0.1, label="Creativity", info="Higher = more creative responses" ) max_tokens_slider = gr.Slider( minimum=1000, maximum=8000, value=4000, step=500, label="Response Length" ) # Tool Activity Section with gr.Group(elem_classes="main-section"): gr.Markdown("### 🔧 MCP Tools Activity") tool_log_display = gr.Textbox( label="", value="Tools will be called during analysis...", lines=4, interactive=False, elem_classes="tool-log" ) with gr.Accordion("📚 Available MCP Tools", open=False): tools_md = "" for name, tool in MCP_TOOLS.items(): tools_md += f"**{name}**\n{tool['description']}\n\n" gr.Markdown(tools_md) with gr.Accordion("🎭 Persona Guide", open=False): for name, desc in PERSONA_DESCRIPTIONS.items(): gr.Markdown(f"**{name}**\n{desc}\n") with gr.Accordion("🔑 API Keys", open=False): gr.Markdown(""" **Get your API key:** • [Anthropic Console](https://console.anthropic.com/) • [OpenAI Platform](https://platform.openai.com/api-keys) • [Google AI Studio](https://aistudio.google.com/app/apikey) **HuggingFace Spaces:** Set in Settings → Secrets: `ANTHROPIC_API_KEY`, `OPENAI_API_KEY`, `GOOGLE_API_KEY` """) clear_btn = gr.Button("🗑️ Clear Chat", variant="secondary") with gr.Column(scale=2): chatbot = gr.Chatbot( label="Conversation", height=520 ) user_input = gr.Textbox( label="Your Idea", placeholder="Share your idea, decision, or question to get challenged...", lines=3, show_label=False ) submit_btn = gr.Button("🚀 Get Second Opinion", variant="primary") gr.Markdown( """
💡 Pro Tips: Be specific about your reasoning • Share your assumptions • The agent will automatically select and use relevant analysis tools • Try different personas for different perspectives
""" ) # Example scenarios gr.Markdown("### 📋 Try an Example") with gr.Row(): example_1 = gr.Button("🏢 Career Move", size="sm") example_2 = gr.Button("💼 Business Pivot", size="sm") example_3 = gr.Button("🎯 Product Launch", size="sm") example_4 = gr.Button("📊 Investment", size="sm") # Event handlers def update_model_choices(provider): provider_config = LLM_PROVIDERS.get(provider, LLM_PROVIDERS["Google (Gemini)"]) return gr.Dropdown( choices=provider_config["models"], value=provider_config["default_model"] ) provider_dropdown.change( fn=update_model_choices, inputs=[provider_dropdown], outputs=[model_dropdown] ) def update_persona_description(persona): return f"
💡 {PERSONA_DESCRIPTIONS[persona]}
" persona_dropdown.change( fn=update_persona_description, inputs=[persona_dropdown], outputs=[persona_description] ) def chat_interaction(user_msg, persona, provider, model, api_key, history, temp, max_tok): if not user_msg.strip(): return format_history_for_gradio(normalize_history(history)), "", "No input provided" response, updated_history, tool_log = get_second_opinion( user_msg, persona, provider, model, api_key, history, temperature=temp, max_tokens=max_tok ) if response.startswith("⚠️"): # Error occurred - append error to history error_history = normalize_history(history) error_history.append({"role": "user", "content": user_msg}) error_history.append({"role": "assistant", "content": response}) return format_history_for_gradio(error_history), "", tool_log or "Error occurred" return updated_history, "", tool_log submit_btn.click( fn=chat_interaction, inputs=[user_input, persona_dropdown, provider_dropdown, model_dropdown, api_key_input, chatbot, temperature_slider, max_tokens_slider], outputs=[chatbot, user_input, tool_log_display] ) user_input.submit( fn=chat_interaction, inputs=[user_input, persona_dropdown, provider_dropdown, model_dropdown, api_key_input, chatbot, temperature_slider, max_tokens_slider], outputs=[chatbot, user_input, tool_log_display] ) def clear_conversation(): return [], "", "Tools will be called during analysis..." clear_btn.click( fn=clear_conversation, outputs=[chatbot, user_input, tool_log_display] ) # Example scenarios def load_example_1(): return "I'm considering leaving my stable job at a big tech company to join an early-stage startup. The startup is in an exciting space (AI tools) and I'd get equity, but the pay is 30% less. I'm 28, single, and have $50k in savings. I think this is the right move for my career growth." def load_example_2(): return "Our SaaS company is planning to move upmarket from small businesses to enterprise customers. We'll increase prices 5x and add enterprise features. This should increase revenue per customer significantly. We have 2,000 SMB customers currently and strong product-market fit." def load_example_3(): return "We want to add AI-powered auto-complete to our text editor. Users have been requesting it, our competitors have it, and we have the technical capability. We're planning to make it enabled by default for all users with an option to turn it off. This seems like a no-brainer feature to add." def load_example_4(): return "I'm thinking of putting 30% of my portfolio into Bitcoin and cryptocurrency. Traditional markets are overvalued, inflation is high, and crypto is the future of finance. I'm 35 and have a moderate risk tolerance. The potential upside seems huge compared to bonds or index funds." example_1.click(fn=load_example_1, outputs=[user_input]) example_2.click(fn=load_example_2, outputs=[user_input]) example_3.click(fn=load_example_3, outputs=[user_input]) example_4.click(fn=load_example_4, outputs=[user_input]) return app # ============================================================================= # MAIN ENTRY POINT # ============================================================================= if __name__ == "__main__": app = create_interface() app.launch()