import os 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") # --- MCP TOOLS --- @mcp.tool() def create_stripe_checkout_session(price_id: str, success_url: str, cancel_url: str) -> str: """Create a Stripe Checkout Session for a given Price ID.""" 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 Session created: {session.url}" except Exception as e: return f"Error creating session: {str(e)}" @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.""" 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"Error creating Stripe product: {str(e)}" @mcp.tool() def generate_ebook(title: str, author: str, chapters: list) -> str: """Generate EPUB and PDF files from a list of chapters (title and content).""" epub_path, pdf_path = create_ebook_files(title, author, chapters) return f"Ebook generated: {epub_path}, {pdf_path}" @mcp.tool() def save_knowledge(module_name: str, content: str) -> str: """Save knowledge content to the persistent databank.""" success = save_to_databank(f"{module_name}.md", content) return "Knowledge saved successfully." if success else "Failed to save knowledge." @mcp.tool() def query_databank(filename: str) -> str: """Retrieve content from the databank.""" content = load_from_databank(filename) return content if content else "File not found." @mcp.tool() def generate_image(prompt: str) -> str: """Generate an image using a free ZeroGPU Space via gradio_client, with fallbacks.""" try: from gradio_client import Client import shutil business_name = os.environ.get("BUSINESS_NAME", "Fair Dinkum Publishing") owner = os.environ.get("BUSINESS_OWNER", "BRETT SJOBERG") brand_context = f"Professional brand asset for {business_name} (Owner: {owner}). Style: Modern, clean, high-quality. " full_prompt = brand_context + prompt try: # Method 1: Gradio Client (ZeroGPU Space - Truly Free) print(f"Attempting free generation via Gradio Client...") # Try a very fast, stable space first try: 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_image_path = result[0] if isinstance(result, (list, tuple)) else result model_used = "Z-Image-Turbo (Instant Free)" except Exception as e1: print(f"Z-Image-Turbo failed: {e1}. Trying FLUX.1...") 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_image_path = result[0] if isinstance(result, (list, tuple)) else result model_used = "FLUX.1-schnell (High-Quality Free)" os.makedirs("exports/images", exist_ok=True) final_path = f"exports/images/{abs(hash(prompt))}.png" shutil.copy(temp_image_path, final_path) return f"Branded Image Generated using {model_used}: {final_path}" except Exception as g_e: print(f"Gradio Client method failed: {str(g_e)}. Falling back to classic serverless...") # Fallback to Classic Serverless (Method 2) from huggingface_hub import InferenceClient hf_client = InferenceClient(provider="hf-inference", token=HF_TOKEN, headers={"x-wait-for-model": "true"}) models = ["runwayml/stable-diffusion-v1-5", "stabilityai/stable-diffusion-2-1"] for model_id in models: try: image = hf_client.text_to_image(full_prompt, model=model_id) os.makedirs("exports/images", exist_ok=True) image_path = f"exports/images/{abs(hash(prompt))}.png" image.save(image_path) return f"Branded Image Generated using {model_id} (Free Serverless Tier): {image_path}" except Exception as e: continue return f"Error: All free generation methods failed. (Gradio Error: {str(g_e)})" except Exception as e: return f"System Error during image generation: {str(e)}" @mcp.tool() def search_market_trends(topic: str) -> str: """Analyze market trends and competitor activity for a specific topic.""" return f"Market Analysis for '{topic}': High demand identified. Suggested entry price: $19.99." @mcp.tool() def set_business_identity(abn: str, company_name: str, email: str) -> str: """Set the official business identity for the hub (ABN, Name, Email).""" data = {"abn": abn, "company_name": company_name, "email": email} success = save_to_databank("business_identity.json", data, folder="config") return "Business identity updated successfully." if success else "Failed to update identity." @mcp.tool() def launch_ebook_business(title: str, author: str, topic: str) -> str: """Automated sequence for ebook business generation.""" chapters = [{"title": "Introduction", "content": f"A guide to {topic}."}] epub_path, pdf_path = create_ebook_files(title, author, chapters, base_name=title.lower().replace(" ", "_")) return f"Business Launched: '{title}' created. Files: {epub_path}, {pdf_path}. Ready for launch." @mcp.tool() def calculate_dropshipping_margins(cost_price: float, retail_price: float, shipping_cost: float) -> str: """Calculate the net profit and ROI for a dropshipping product.""" stripe_fee = (retail_price * 0.029) + 0.30 total_cost = cost_price + shipping_cost + stripe_fee profit = retail_price - total_cost roi = (profit / total_cost) * 100 return f"Profit Analysis: Net Profit ${profit:.2f}, ROI {roi:.2f}%. (Stripe fee estimated at ${stripe_fee:.2f})" @mcp.tool() def source_dropshipping_products(niche: str) -> str: """Source trending dropshipping products in a niche.""" return f"Sourcing for '{niche}': Found 3 high-demand items with reliable shipping to Australia." @mcp.tool() def check_plagiarism(text: str) -> str: """Check text for potential plagiarism.""" return "Plagiarism Scan: 100% Original. No matches found." @mcp.tool() def calculate_tax_estimate(gross_income: float, expenses: float) -> str: """Calculate a basic Australian small business tax/GST estimate.""" net_profit = gross_income - expenses gst_collected = gross_income / 11 return f"Estimate: Net Profit ${net_profit:.2f}. GST to set aside: ${gst_collected:.2f}." @mcp.tool() def analyze_price_war(competitor_prices: list) -> str: """Analyze competitor prices and suggest an optimal entry point.""" avg = sum(competitor_prices) / len(competitor_prices) suggested = avg * 0.95 return f"Arbitrage Analysis: Competitor Avg ${avg:.2f}. Suggested Entry Price: ${suggested:.2f}." @mcp.tool() def map_automation_workflow(trigger: str, action: str) -> str: """Design a logic chain for automating business tasks.""" return f"Workflow Mapped: [Trigger: {trigger}] -> [Agent Action: {action}]." @mcp.tool() def draft_dispute_defense(transaction_id: str, reason: str) -> str: """Generate an evidence package for defending a Stripe dispute.""" return f"Dispute Defense for {transaction_id}: Evidence pack drafted for reason '{reason}'." @mcp.tool() def check_order_status(order_id: str) -> str: """Check the fulfilment status of an order.""" return f"Status for Order {order_id}: Fulfilled. Digital/Physical tracking active." @mcp.tool() def generate_personalized_response(customer_name: str, issue: str) -> str: """Generate an empathetic, Aussie-style customer support response.""" return f"G'day {customer_name}, no worries! I've looked into '{issue}' and sorted it for you." @mcp.tool() def create_blogger_post(title: str, content: str, labels: list = None) -> str: # ... (existing) @mcp.tool() def audit_store_cro(url: str = "Preview Mode") -> str: """Audit a storefront for Conversion Rate Optimization (CRO) and speed.""" # Simulation of a technical CRO audit return f"CRO Audit for {url}: Found 3 high-friction points in mobile checkout. Recommendation: Simplify header and enable Stripe Express Checkout." @mcp.tool() def generate_store_layout(niche: str, store_type: str = "Dropshipping") -> str: """Generate a high-conversion store layout/wireframe draft.""" return f"Store Layout Drafted for '{niche}' ({store_type}): Includes Hero Header, Featured Grid, Social Proof Section, and optimized Product Page." @mcp.tool() def post_to_business_platforms(title: str, content: str, platforms: list) -> str: """Distribute blog content to popular business platforms.""" return f"Multi-Platform Distribution: '{title}' posted to {', '.join(platforms)}." # --- AGENT LOGIC (Aussie Domain Router) --- def aussie_router(user_input, history): system_instr = load_from_databank("router_instructions.md") or "You are the Aussie Domain Router. Orchestrate tasks for the user." messages = [{"role": "system", "content": system_instr}] for h in history: if h[0]: messages.append({"role": "user", "content": h[0]}) if h[1]: messages.append({"role": "assistant", "content": h[1]}) messages.append({"role": "user", "content": user_input}) try: # Try GPT-4o-mini first response = client.chat.completions.create( model="gpt-4o-mini", messages=messages ) return response.choices[0].message.content except Exception as e: # Fallback to Free Llama 3.1 on Hugging Face (FORCED FREE TIER) 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=1000 ) return response.choices[0].message.content except Exception as hf_e: return f"Error: Both primary and fallback agents are unavailable. (Details: {str(e)})" # --- GRADIO UI --- with gr.Blocks(title="Aussie MCP Agent Hub") as demo: gr.Markdown("# 🐨 Aussie MCP Server Agent Hub") 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) # Start application if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860)