MCP Agentic AI Server

import os import uuid import logging from google import genai from custom_mcp.tools.sample_tool import sample_tool # ——— Instantiate Gemini client with explicit API key ——— api_key = os.getenv("GEMINI_API_KEY") if not api_key: logging.error("GEMINI_API_KEY is not set in the environment!") raise RuntimeError("Missing GEMINI_API_KEY") client = genai.Client(api_key=api_key) logging.info("Initialized Gemini Client with provided API key") import threading import time class MCPController: def __init__(self): self.tasks = {} self.queries_processed = 0 self.total_response_time = 0.0 self.successful_queries = 0 self.failed_queries = 0 self.session_start_time = time.time() self.lock = threading.Lock() logging.info("MCPController initialized") def create_task(self, user_input: str, tools: list[str]) -> str: task_id = str(uuid.uuid4()) self.tasks[task_id] = {"input": user_input, "tools": tools} logging.info("Created task %s input=%r tools=%s", task_id, user_input, tools) return task_id def run(self, task_id: str) -> dict: if task_id not in self.tasks: logging.error("Task %s not found", task_id) with self.lock: self.failed_queries += 1 return {"error": "Task not found"} task = self.tasks[task_id] text = task["input"] if "sample_tool" in task["tools"]: text = sample_tool(text) logging.info("After sample_tool: %r", text) prompt = f"Process the input: {text}" logging.info("Sending to Gemini: %s", prompt) start_time = time.time() try: resp = client.models.generate_content( model="gemini-2.5-flash", contents=prompt ) output = resp.text logging.info("Gemini output: %r", output) with self.lock: self.queries_processed += 1 self.successful_queries += 1 self.total_response_time += (time.time() - start_time) except Exception as e: logging.exception("Gemini call failed") with self.lock: self.queries_processed += 1 self.failed_queries += 1 self.total_response_time += (time.time() - start_time) return {"task_id": task_id, "error": str(e)} return {"task_id": task_id, "output": output} def get_stats(self): with self.lock: elapsed = time.time() - self.session_start_time avg_response = (self.total_response_time / self.queries_processed) if self.queries_processed else 0.0 success_rate = (self.successful_queries / self.queries_processed * 100) if self.queries_processed else 0.0 return { "active_sessions": 1, # For demo, 1 process = 1 session "queries_processed": self.queries_processed, "response_time": round(avg_response, 2), "success_rate": round(success_rate, 2), "todays_queries": self.queries_processed, # For demo, all queries are today's "uptime": round(elapsed / 60, 2) # Uptime in minutes }