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 with multiple model fallbacks using the free hf-inference provider.""" try: from huggingface_hub import InferenceClient hf_client = InferenceClient( provider="hf-inference", token=HF_TOKEN, headers={"x-wait-for-model": "true"} ) 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 # Truly free models on hf-inference models = [ "stabilityai/stable-diffusion-xl-base-1.0", "runwayml/stable-diffusion-v1-5", "CompVis/stable-diffusion-v1-4" ] image = None model_used = None for model_id in models: try: image = hf_client.text_to_image(full_prompt, model=model_id) model_used = model_id break except Exception as e: print(f"Model {model_id} failed: {str(e)}") continue if not image: return "Error: All free image generation models failed. The serverless API may be overloaded." 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_used} (Free Tier): {image_path}" 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: """Draft a new blog post for Fair Dinkum Publishing on Blogger.""" return f"Blog Post Drafted: '{title}'. Content queued for publishing on Blogger." @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)