HubSpot Dynamic Multi-Agent System with Magistral Reasoning

import requests
import json
from mistralai import Mistral, ThinkChunk, TextChunk
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional
import re

HUBSPOT_API_KEY = "<YOUR HUBSPOT API KEY>"  # Replace with your HubSpot API key
MISTRAL_API_KEY = "<YOUR MISTRAL API KEY>"  # Get it from https://console.mistral.ai/api-keys

mistral_client = Mistral(api_key=MISTRAL_API_KEY)

class HubSpotConnector:
    """Handles all HubSpot API operations"""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.hubapi.com/crm/v3/objects"
        self.headers = {"Authorization": f"Bearer {api_key}"}

    def get_properties(self) -> Dict:
        """Load all HubSpot properties"""
        print("📡 HubSpotConnector: Loading properties...")
        properties = {}

        for obj_type in ['deals', 'contacts', 'companies']:
            url = f"https://api.hubapi.com/crm/v3/properties/{obj_type}"
            response = requests.get(url, headers=self.headers)

            if response.status_code == 200:
                props = response.json()['results']
                prop_list = []
                for prop in sorted(props, key=lambda x: x['name']):
                    prop_str = f"'{prop['name']}' - {prop['label']}"
                    if 'options' in prop and prop['options']:
                        valid_values = [opt['value'] for opt in prop['options']]
                        prop_str += f" | Valid values: {valid_values}"
                    prop_list.append(prop_str)

                properties[obj_type] = prop_list

        print(f"✅ HubSpotConnector: Loaded properties for {len(properties)} object types")
        return properties

    def get_data(self, object_type: str) -> List[Dict]:
        """Fetch data from HubSpot"""
        print(f"📡 HubSpotConnector: Fetching {object_type} data...")

        url = f"{self.base_url}/{object_type}"
        params = {"limit": 100}
        all_data = []

        while url:
            response = requests.get(url, headers=self.headers, params=params)
            if response.status_code != 200:
                raise Exception(f"HubSpot API error: {response.text}")

            data = response.json()
            all_data.extend(data.get("results", []))
            url = data.get("paging", {}).get("next", {}).get("link")
            params = {}

        print(f"✅ HubSpotConnector: Loaded {len(all_data)} {object_type}")
        return all_data

    def batch_update(self, updates: Dict) -> None:
        """Perform batch updates to HubSpot"""
        for object_type, update_list in updates.items():
            if not update_list:
                continue

            print(f"📡 HubSpotConnector: Updating {len(update_list)} {object_type}...")

            url = f"{self.base_url}/{object_type}/batch/update"
            headers = {**self.headers, "Content-Type": "application/json"}

            # Process in batches of 100
            for i in range(0, len(update_list), 100):
                batch = update_list[i:i+100]
                payload = {"inputs": batch}

                response = requests.post(url, headers=headers, json=payload)
                if response.status_code not in [200, 202]:
                    raise Exception(f"HubSpot update error: {response.text}")

            print(f"✅ HubSpotConnector: {object_type} updates completed")

def magistral_reasoning(prompt: str) -> Dict[str, str]:
    """Use reasoning model for query analysis and planning"""
    response = mistral_client.chat.complete(
        model="magistral-medium-latest",
        messages=[{"role": "user", "content": prompt}]
    )

    content = response.choices[0].message.content

    reasoning = ""
    conclusion = ""

    for r in content:
      if isinstance(r, ThinkChunk):
          reasoning = r.thinking[0].text
      elif isinstance(r, TextChunk):
          conclusion = r.text

    return {
        "reasoning": reasoning,
        "conclusion": conclusion
    }

def mistral_small_execution(prompt: str) -> str:
    """Use Mistral Small for content generation"""
    response = mistral_client.chat.complete(
        model="mistral-small-latest",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

class LeadAgent:
    """Lead Agent powered by Magistral reasoning model for query analysis and planning"""

    def __init__(self, hubspot_properties):
        self.hubspot_properties = hubspot_properties
        self.name = "LeadAgent"

    def analyze_query(self, query: str) -> Dict:
        """Analyze query using Magistral reasoning and create execution plan"""
        print(f"🧠 {self.name}: Analyzing query with Magistral reasoning...")

        analysis_prompt = f"""
        Analyze this HubSpot query and create a detailed execution plan based on different hubspot properties provided by following the shared rules:

        HUBSPOT_PROPERTIES: {self.hubspot_properties}

        QUERY: {query}

        RULES:
        1. What is the user asking for?
        2. Is this a read-only query or does it require HubSpot updates?
        3. What sub-agents are needed to accomplish this?
        4. What HubSpot data is required?
        5. What's the execution sequence?
        6. What should be the final output format?
        7. Query can also be combination of read-only and write-back.
        8. Query is read-only if it requires data read from HubSpot.
        9. Query is write-back if it requires an update to existing values or writing/ assigning new values.
        10. In the final conclusion just give only one JSON string nothing else. I don't need any explanation.

        Provide a JSON execution plan with:
        {{
            "sub_agents": [
                {{
                    "name": "agent_name",
                    "task": "specific task description",
                    "task_type": "read_only" or "write_back",
                    "input_data": ["deals", "contacts", "companies"],
                    "output_format": "expected output"
                }}
            ]
        }}
        """

        # Use existing magistral_reasoning function
        analysis = magistral_reasoning(analysis_prompt)

        try:
            # Extract JSON execution plan from conclusion
            json_match = re.search(r'\{.*\}', analysis["conclusion"], re.DOTALL)
            if json_match:
                execution_plan = json.loads(json_match.group(0))
            else:
                raise ValueError("No JSON found in analysis")
        except Exception as e:
            print(f"⚠️ {self.name}: JSON parsing failed, using fallback plan")
            execution_plan = {
                "sub_agents": [{
                    "name": "general_analyzer",
                    "task": query,
                    "task_type": "read_only",
                    "input_data": ["deals", "contacts", "companies"],
                    "output_format": "summary"
                }]
            }

        print(f"✅ {self.name}: Plan created - {len(execution_plan['sub_agents'])} sub-agents needed")

        return {
            "reasoning": analysis["reasoning"],
            "execution_plan": execution_plan,
            "conclusion": analysis["conclusion"]
        }

class SubAgent:
    """Dynamic Sub-Agent created on-the-fly for specific tasks"""

    def __init__(self, name: str, task: str, task_type: str, input_data: List[str],
                 output_format: str):
        self.name = name
        self.task = task
        self.task_type = task_type
        self.input_data = input_data
        self.output_format = output_format

    def execute(self, data: Dict, properties_context: str, hubspot_updater=None) -> Dict:
        """Execute the assigned task"""
        print(f"🤖 {self.name} ({self.task_type}): Executing task...")

        if self.task_type == 'read_only':
            agent_prompt = f"""
            You are a {self.name} agent.

            TASK: {self.task}

            AVAILABLE HUBSPOT PROPERTIES:
            {properties_context}

            DATA AVAILABLE:
            {json.dumps(data, indent=2)}

            OUTPUT FORMAT: {self.output_format}

            Provide your analysis only based on the available data.
            """
        else:  # write_back
            agent_prompt = f"""
            You are a {self.name} agent.

            TASK: {self.task}

            AVAILABLE HUBSPOT PROPERTIES:
            {properties_context}

            DATA AVAILABLE:
            {json.dumps(data, indent=2)}

            CRITICAL: Use exact HubSpot property names from the list above in your JSON output.

            OUTPUT FORMAT: JSON format with the properties to be written to HubSpot

            Provide updates using exact HubSpot property names.
            """

        # Use existing mistral_small_execution function
        result = mistral_small_execution(agent_prompt)

        # Handle write-back operations
        if self.task_type == 'write_back' and hubspot_updater:
            try:
                json_match = re.search(r'\{.*\}', result, re.DOTALL)
                if json_match:
                    updates = json.loads(json_match.group(0))
                    hubspot_updater.batch_update(updates)
                    print(f"✅ {self.name}: Successfully updated HubSpot records")
            except Exception as e:
                print(f"❌ {self.name}: Update failed - {str(e)}")
                return {"status": "error", "error": str(e), "raw_result": result}

        print(f"✅ {self.name}: Task completed successfully")
        return {"status": "success", "result": result}

class SynthesisAgent:
    """Final agent to synthesize all results into user-friendly response"""

    def __init__(self):
        self.name = "SynthesisAgent"

    def synthesize(self, query: str, sub_agent_results: List[Dict], execution_plan: Dict) -> str:
        """Combine all sub-agent results into final answer"""
        print(f"🔄 {self.name}: Synthesizing results from {len(sub_agent_results)} agents...")

        # Prepare context from all sub-agent results
        results_context = ""
        for result in sub_agent_results:
            results_context += f"\n{result['agent'].upper()} ({result['task_type']}):\n"
            if result['result']['status'] == 'success':
                results_context += f"{result['result']['result']}\n"
            else:
                results_context += f"Error: {result['result'].get('error', 'Unknown error')}\n"
            results_context += "---\n"

        synthesis_prompt = f"""
        You are a final synthesizer agent. Create a comprehensive, user-friendly response based on all sub-agent results.

        ORIGINAL QUERY: {query}

        SUB-AGENT RESULTS:
        {results_context}

        TASK: Synthesize all the above results into a clear, actionable response for the user.

        Guidelines:
        1. Start with a direct answer to the user's query
        2. Include key insights and findings
        3. If updates were made, summarize what was changed
        4. Provide actionable next steps if relevant
        5. Keep it concise but comprehensive
        6. Use a professional but friendly tone

        Provide the final synthesized response:
        """

        # Use existing mistral_small_execution function
        final_answer = mistral_small_execution(synthesis_prompt)
        print(f"✅ {self.name}: Final answer synthesized")

        return final_answer

class AgentOrchestrator:
    """Main orchestrator that coordinates all agents"""

    def __init__(self, hubspot_api_key: str, mistral_api_key: str):
        # Initialize global mistral client for existing functions
        global mistral_client
        mistral_client = Mistral(api_key=mistral_api_key)

        # Initialize HubSpot connector
        self.hubspot_connector = HubSpotConnector(hubspot_api_key)

        # Load HubSpot data and properties
        self.hubspot_properties = self.hubspot_connector.get_properties()
        self.hubspot_data = {
            "deals": self.hubspot_connector.get_data("deals"),
            "contacts": self.hubspot_connector.get_data("contacts"),
            "companies": self.hubspot_connector.get_data("companies")
        }

        # Initialize agents
        self.lead_agent = LeadAgent(self.hubspot_properties)
        self.synthesis_agent = SynthesisAgent()
        self.active_sub_agents = []

        print(f"🚀 AgentOrchestrator: System initialized with {sum(len(data) for data in self.hubspot_data.values())} HubSpot records")

    def process_query(self, query: str) -> Dict:
        """Main method to process user queries through multi-agent workflow"""
        print(f"\n🎯 Processing Query: {query[:100]}...")
        print("=" * 80)

        # Step 1: Lead Agent analyzes query using Magistral reasoning
        analysis = self.lead_agent.analyze_query(query)
        execution_plan = analysis["execution_plan"]

        # Step 2: Create and execute sub-agents dynamically
        sub_agent_results = []
        self.active_sub_agents = []

        for agent_config in execution_plan["sub_agents"]:
            # Create sub-agent dynamically
            sub_agent = SubAgent(
                name=agent_config["name"],
                task=agent_config["task"],
                task_type=agent_config["task_type"],
                input_data=agent_config["input_data"],
                output_format=agent_config["output_format"]
            )

            self.active_sub_agents.append(sub_agent)

            # Prepare data and context for this sub-agent
            agent_data = {data_type: self.hubspot_data.get(data_type, [])
                         for data_type in agent_config["input_data"]}

            # Build properties context
            properties_context = ""
            for data_type in agent_config["input_data"]:
                if data_type in self.hubspot_properties:
                    properties_context += f"\n{data_type.upper()} PROPERTIES:\n"
                    properties_context += "\n".join(self.hubspot_properties[data_type])
                    properties_context += "\n"

            # Execute sub-agent using mistral_small_execution
            result = sub_agent.execute(agent_data, properties_context, self.hubspot_connector)
            sub_agent_results.append({
                "agent": sub_agent.name,
                "task_type": sub_agent.task_type,
                "result": result
            })

        # Step 3: Synthesis Agent creates final answer using mistral_small_execution
        final_answer = self.synthesis_agent.synthesize(query, sub_agent_results, execution_plan)

        print("=" * 80)
        print("✨ Query processing completed!")

        return {
            "query": query,
            "reasoning": analysis["reasoning"],
            "execution_plan": execution_plan,
            "sub_agent_results": sub_agent_results,
            "active_agents": [agent.name for agent in self.active_sub_agents],
            "final_answer": final_answer
        }