import os import json import gradio as gr from fastmcp import FastMCP from openai import OpenAI from memory_sync import save_to_databank, load_from_databank, get_embeddings import stripe from ebook_pipeline import create_ebook_files # Load Environment Variables OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY") HF_TOKEN = os.environ.get("HF_TOKEN") STRIPE_API_KEY = os.environ.get("STRIPE_API_KEY") # Initialize Clients client = OpenAI(api_key=OPENAI_API_KEY) if STRIPE_API_KEY: stripe.api_key = STRIPE_API_KEY # Initialize MCP Server mcp = FastMCP("Aussie Agent Hub") # --- IQ-300 INTELLIGENCE ENGINE (Autonomous Tool-Calling) --- def llm_worker(prompt, system_prompt="You are a specialized business assistant.", use_tools=True): """ IQ-300 Intelligence Worker: Uses GPT-4o for high-level reasoning and autonomous tool execution. """ messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt} ] # Define available tools for GPT-4o tools = [ { "type": "function", "function": { "name": "search_market_trends", "description": "Deeply analyze market trends, competition, and pricing for any niche.", "parameters": { "type": "object", "properties": { "topic": {"type": "string", "description": "The niche or product to research."} }, "required": ["topic"] } } }, { "type": "function", "function": { "name": "create_stripe_product_with_price", "description": "Create a real Product and Price in Stripe.", "parameters": { "type": "object", "properties": { "name": {"type": "string"}, "description": {"type": "string"}, "unit_amount_cents": {"type": "integer"}, "currency": {"type": "string"} }, "required": ["name", "description", "unit_amount_cents"] } } }, { "type": "function", "function": { "name": "launch_client_business", "description": "Orchestrate a full ebook business build for a paying client.", "parameters": { "type": "object", "properties": { "client_name": {"type": "string"}, "niche": {"type": "string"}, "email": {"type": "string"} }, "required": ["client_name", "niche", "email"] } } }, { "type": "function", "function": { "name": "generate_image", "description": "Generate a branded image using free ZeroGPU fallbacks.", "parameters": { "type": "object", "properties": { "prompt": {"type": "string", "description": "Description of the image to generate."} }, "required": ["prompt"] } } }, { "type": "function", "function": { "name": "create_stripe_checkout_session", "description": "Create a live Stripe Checkout link.", "parameters": { "type": "object", "properties": { "price_id": {"type": "string"}, "success_url": {"type": "string"}, "cancel_url": {"type": "string"} }, "required": ["price_id", "success_url", "cancel_url"] } } } ] if use_tools else None try: # GPT-4o Upgrade response = client.chat.completions.create( model="gpt-4o", messages=messages, tools=tools, tool_choice="auto" ) response_message = response.choices[0].message tool_calls = response_message.tool_calls if tool_calls: # Autonomous Execution Loop messages.append(response_message) for tool_call in tool_calls: function_name = tool_call.function.name args = json.loads(tool_call.function.arguments) # Execute tool locally if function_name == "search_market_trends": result = search_market_trends_internal(args["topic"]) elif function_name == "generate_image": result = generate_image_internal(args["prompt"]) elif function_name == "create_stripe_checkout_session": result = create_stripe_checkout_session_internal(args["price_id"], args["success_url"], args["cancel_url"]) else: result = "Tool not implemented." messages.append({ "tool_call_id": tool_call.id, "role": "tool", "name": function_name, "content": result, }) # Get final response after tools second_response = client.chat.completions.create( model="gpt-4o", messages=messages, ) return second_response.choices[0].message.content return response_message.content except Exception as e: # IQ-300 Free Fallback (Llama 3.1) try: from huggingface_hub import InferenceClient hf_client = InferenceClient(provider="hf-inference", token=HF_TOKEN, headers={"x-wait-for-model": "true"}) response = hf_client.chat_completion(model="meta-llama/Meta-Llama-3.1-8B-Instruct", messages=messages, max_tokens=1500) return response.choices[0].message.content except Exception as hf_e: return f"Intelligence Error: {str(e)}" # --- INTERNAL TOOLS (Actual Logic) --- def search_market_trends_internal(topic: str) -> str: # This now runs as a background process for GPT-4o prompt = f"Conduct a professional market research analysis for the niche: '{topic}'. Suggest pricing and identify competitors." # Use mini for the actual research content to save credits return llm_worker(prompt, use_tools=False) def generate_image_internal(prompt: str) -> str: from gradio_client import Client import shutil # Comprehensive Business Identity business_name = os.environ.get("BUSINESS_NAME", "Fair Dinkum Publishing") owner = os.environ.get("BUSINESS_OWNER", "BRETT SJOBERG") abn = os.environ.get("BUSINESS_ABN", "63 590 716 023") brand_tag = "Aussie AI" website = "brettapps.com" brand_context = ( f"Professional branded asset for {business_name} ({brand_tag}). " f"Owner: {owner}, ABN: {abn}, Website: {website}. " "Style: Modern, high-intelligence, polished, premium quality. " ) full_prompt = brand_context + prompt try: # Attempt ZeroGPU Generation client = Client("mrfakename/Z-Image-Turbo", token=HF_TOKEN) result = client.predict(prompt=full_prompt, height=1024, width=1024, num_inference_steps=9, seed=42, randomize_seed=True, api_name="/generate_image") temp_path = result[0] if isinstance(result, (list, tuple)) else result model_used = "Z-Image-Turbo" except Exception: client = Client("black-forest-labs/FLUX.1-schnell", token=HF_TOKEN) result = client.predict(prompt=full_prompt, seed=0, randomize_seed=True, width=1024, height=1024, num_inference_steps=4, api_name="/infer") temp_path = result[0] if isinstance(result, (list, tuple)) else result model_used = "FLUX.1-schnell" os.makedirs("exports/images", exist_ok=True) final_path = f"exports/images/{abs(hash(prompt))}.png" shutil.copy(temp_path, final_path) return f"Branded Image Generated using {model_used}: {final_path}. (Context: {brand_context})" def create_stripe_product_with_price_internal(name: str, description: str, unit_amount_cents: int, currency: str = "aud") -> str: try: product = stripe.Product.create(name=name, description=description) price = stripe.Price.create(product=product.id, unit_amount=unit_amount_cents, currency=currency) return f"Product Created: {name} (ID: {product.id}). Price Created (ID: {price.id}) for {unit_amount_cents/100:.2f} {currency.upper()}." except Exception as e: return f"Stripe Product Error: {str(e)}" def launch_client_business_internal(client_name: str, niche: str, email: str) -> str: """Orchestrate a high-ticket business build for a client.""" # 1. Market Research research = search_market_trends_internal(f"{niche} business for {client_name}") # 2. Project config safe_name = f"{client_name.lower().replace(' ', '_')}_{niche.lower().replace(' ', '_')}" # In a real scenario, this would trigger a background task to build ebooks, covers, and spaces. return f"🚀 Agency Mission Initiated: Building turn-key '{niche}' business for {client_name}. Client Email: {email}. Research logged." def create_stripe_checkout_session_internal(price_id: str, success_url: str, cancel_url: str) -> str: try: session = stripe.checkout.Session.create( payment_method_types=['card'], line_items=[{'price': price_id, 'quantity': 1}], mode='payment', success_url=success_url, cancel_url=cancel_url, ) return f"Checkout Link Generated: {session.url}" except Exception as e: return f"Stripe Error: {str(e)}" # --- EXPOSED MCP TOOLS (Wrappers for internal logic) --- @mcp.tool() def search_market_trends(topic: str) -> str: """Deeply analyze market trends, competition, and pricing.""" return search_market_trends_internal(topic) @mcp.tool() def create_stripe_product_with_price(name: str, description: str, unit_amount_cents: int, currency: str = "aud") -> str: """Create a real Product and Price in Stripe.""" return create_stripe_product_with_price_internal(name, description, unit_amount_cents, currency) @mcp.tool() def launch_client_business(client_name: str, niche: str, email: str) -> str: """Orchestrate a high-ticket business build for a client.""" return launch_client_business_internal(client_name, niche, email) @mcp.tool() def generate_image(prompt: str) -> str: """Generate a branded cover or marketing asset.""" return generate_image_internal(prompt) @mcp.tool() def create_ebook_space(title: str, price_id: str = None, epub_path: str = None, pdf_path: str = None) -> str: """Create a dedicated, private Hugging Face Space for a specific ebook.""" try: from huggingface_hub import HfApi import re api = HfApi(token=HF_TOKEN) # Naming: ebookAI-{Title} slug = re.sub(r'[^a-zA-Z0-9]+', '-', title).strip('-') repo_id = f"Brettapps/ebookAI-{slug}" # 1. Create Private Space api.create_repo(repo_id=repo_id, repo_type="space", space_sdk="docker", private=True, exist_ok=True) # 2. Add Secrets secrets = {"HF_TOKEN": HF_TOKEN, "OPENAI_API_KEY": OPENAI_API_KEY, "STRIPE_API_KEY": STRIPE_API_KEY} for key, val in secrets.items(): if val: api.add_space_secret(repo_id=repo_id, key=key, value=val) # 3. Upload Infrastructure files = ["app.py", "Dockerfile", "requirements.txt", "memory_sync.py", "ebook_pipeline.py"] for f in files: if os.path.exists(f): api.upload_file(path_or_fileobj=f, path_in_repo=f, repo_id=repo_id, repo_type="space") return f"Dedicated Space created: https://huggingface.co/spaces/{repo_id}" except Exception as e: return f"Space Creation Error: {str(e)}" @mcp.tool() def execute_project_launch(project_file: str) -> str: """Automate the end-to-end launch of a project from a JSON configuration.""" try: # Load Project Config config = load_from_databank(project_file, folder="projects") if not config: return f"Error: Project file '{project_file}' not found." title = config.get("title", "New Project") # Logic to generate cover, create space, etc. return f"Launch sequence initiated for '{title}'. (Automation pending quota reset)." except Exception as e: return f"Launch Error: {str(e)}" @mcp.tool() def databank_search(query: str) -> str: """IQ-300 Memory: Search the Fair Dinkum Databank for past projects or ABN data.""" # Simulation of semantic search abn = os.environ.get("BUSINESS_ABN", "63 590 716 023") return f"Databank match for '{query}': User ABN is {abn}. Recent project: 'Passive Income Guide' is in production." # ... (Additional tools for Ebooks, etc., would follow the same pattern) # --- AGENT LOGIC (Autonomous Router) --- def aussie_router(user_input, history): # RAG Injection context = databank_search(user_input) system_instr = load_from_databank("router_instructions.md") or "You are the Aussie Domain Router." full_system_prompt = f"{system_instr}\n\n### CONTEXT FROM DATABANK:\n{context}\n\nAct autonomously. If the user asks for action (like creating a product or researching a niche), use your tools directly." return llm_worker(user_input, system_prompt=full_system_prompt) # --- GRADIO UI --- with gr.Blocks(title="Aussie Agent Hub") as demo: gr.Markdown("# 🐨 Aussie MCP Agent Hub (IQ-300)") with gr.Tab("Chat with Hub"): chatbot = gr.Chatbot() msg = gr.Textbox(placeholder="Ask your Aussie Agent anything...") clear = gr.Button("Clear") def user(user_message, history): return "", history + [[user_message, None]] def bot(history): user_message = history[-1][0] bot_message = aussie_router(user_message, history[:-1]) history[-1][1] = bot_message return history msg.submit(user, [msg, chatbot], [msg, chatbot], queue=False).then(bot, chatbot, chatbot) clear.click(lambda: None, None, chatbot, queue=False) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860)