push

.gitignore

@@ -0,0 +1,44 @@
+# Byte-code files
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+dist/
+*.egg-info/
+*.egg
+
+# Editors
+.idea/
+.vscode/
+
+# Environments
+.venv/
+env/
+venv/
+*.ini
+.python-version
+
+# Logs and databases
+*.log
+*.sqlite3
+*.db
+
+# OS
+.DS_Store
+.Trashes
+Thumbs.db
+
+# uv specific
+.uv/
+
+# dotenv
+.env
+.env.*
+!.env.example # Keep example files

Dockerfile

@@ -0,0 +1,17 @@
+FROM python:3.10
+# Pehle root user hai by default, to yahan packages install karo
+RUN apt-get update && apt-get install -y libgl1 libglib2.0-0
+
+# Ab user add karo aur switch karo
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+
+WORKDIR /app
+
+COPY --chown=user ./requirements.txt requirements.txt
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+RUN pip install --no-cache-dir openai-chatkit==1.4.0
+
+COPY --chown=user . /app
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

ai_mcp_server.py

@@ -0,0 +1,229 @@
+"""
+
+HTTP-only MCP Server for the separated AI service.
+Communicates with the business logic service via HTTP API calls.
+"""
+import re
+from typing import Literal, Optional, Any, Dict
+import httpx
+from mcp.server.fastmcp import FastMCP
+from config.config import settings
+import logging
+import os
+import sys
+
+# Set up logging
+logger = logging.getLogger(__name__)
+
+logger.info("--- MCP SERVER SCRIPT STARTING ---")
+# Load secure service secret from environment variable
+SERVICE_SECRET = settings.BETTER_AUTH_SECRET
+logger.info(f"SERVICE_SECRET loaded: '{bool(SERVICE_SECRET)}'") # Log boolean status, not the secret itself.
+
+if not SERVICE_SECRET:
+    logger.error("SERVICE_SECRET is NOT SET. Authentication will fail.")
+
+# Create MCP server instance
+mcp = FastMCP("separated-task-management-server")
+logger.info("FastMCP instance created.")
+
+# Get business service base URL from environment
+BUSINESS_SERVICE_URL = settings.BUSINESS_SERVICE_URL
+
+def _get_auth_headers() -> Dict[str, str]:
+    if not SERVICE_SECRET:
+        logger.error("Attempted to get auth headers but SERVICE_SECRET is not set.")
+        return {}
+    return {"Authorization": f"Bearer {SERVICE_SECRET}"}
+
+def detect_priority_from_text(text: str) -> str:
+    text_lower = text.lower()
+    high_priority_patterns = [r'\bhigh\s*priority\b', r'\burgent\b', r'\bcritical\b', r'\bimportant\b', r'\basap\b', r'\bhigh\b']
+    low_priority_patterns = [r'\blow\s*priority\b', r'\bminor\b', r'\boptional\b', r'\bwhen\s*you\s*have\s*time\b', r'low']
+    if any(re.search(p, text_lower) for p in high_priority_patterns):
+        return "high"
+    if any(re.search(p, text_lower) for p in low_priority_patterns):
+        return "low"
+    if re.search(r'\bmedium\b|\bnormal\b', text_lower):
+        return "medium"
+    return "medium"
+
+@mcp.tool()
+async def add_task(
+    user_id: str,
+    title: str,
+    description: Optional[str] = None,
+    priority: Optional[str] = None,
+    due_date: Optional[str] = None,
+) -> dict:
+    """
+    Create a new task. This tool is idempotent.
+    """
+    if not SERVICE_SECRET:
+        return {"success": False, "done": True, "error": "Internal authentication error: Service secret missing."}
+
+    try:
+        existing_tasks_response = await list_tasks(user_id=user_id)
+        if existing_tasks_response.get("success"):
+            for task in existing_tasks_response.get("tasks", []):
+                if task.get("title", "").lower() == title.lower():
+                    return {"success": False, "done": True, "message": "Task already exists."}
+    except Exception as e:
+        logger.error(f"Failed to check for existing tasks during add_task: {e}")
+        return {"success": False, "done": True, "error": "Failed to verify if task exists.", "details": str(e)}
+
+    if priority is None:
+        combined_text = f"{title} {description or ''}"
+        priority = detect_priority_from_text(combined_text)
+    else:
+        priority = priority.lower()
+        if priority not in ["low", "medium", "high"]:
+            priority = "medium"
+
+    payload = {"title": title, "description": description, "priority": priority, "completed": False, "due_date": due_date}
+    headers = {"Content-Type": "application/json", **_get_auth_headers()}
+
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.post(f"{BUSINESS_SERVICE_URL}/api/{user_id}/tasks", json=payload, headers=headers)
+            if response.status_code in [200, 201]:
+                return {"success": True, "done": True, "message": "Task added successfully."}
+            else:
+                return {"success": False, "done": True, "error": f"Failed to create task: {response.status_code}", "details": response.text}
+        except Exception as e:
+            return {"success": False, "done": True, "error": "Failed to create task.", "details": str(e)}
+
+@mcp.tool()
+async def list_tasks(user_id: str, status: Literal["all", "pending", "completed"] = "all") -> dict:
+    """
+    Retrieve tasks.
+    """
+    if not SERVICE_SECRET:
+        return {"success": False, "done": True, "error": "Internal authentication error: Service secret missing."}
+    params = {"status": status} if status != "all" else {}
+    headers = _get_auth_headers()
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.get(f"{BUSINESS_SERVICE_URL}/api/{user_id}/tasks", params=params, headers=headers)
+            if response.status_code == 200:
+                result = response.json()
+                tasks = result if isinstance(result, list) else result.get("tasks", [])
+                task_list = [{"id": t.get("id"), "title": t.get("title"), "description": t.get("description"), "completed": t.get("completed", False), "priority": t.get("priority"), "created_at": t.get("created_at")} for t in tasks]
+                return {"success": True, "done": True, "tasks": task_list, "count": len(task_list)}
+            else:
+                return {"success": False, "done": True, "error": f"Failed to list tasks: {response.status_code}", "details": response.text}
+        except Exception as e:
+            return {"success": False, "done": True, "error": "Failed to list tasks.", "details": str(e)}
+
+@mcp.tool()
+async def complete_task(user_id: str, task_id: int) -> dict:
+    """
+    Mark a task as complete.
+    """
+    if not SERVICE_SECRET:
+        return {"success": False, "done": True, "error": "Internal authentication error: Service secret missing."}
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.patch(f"{BUSINESS_SERVICE_URL}/api/{user_id}/tasks/{task_id}/complete", json=payload, headers=headers)
+            if response.status_code in [200, 204]:
+                return {"success": True, "done": True, "message": f"Task {task_id} marked as complete."}
+            else:
+                return {"success": False, "done": True, "error": f"Failed to complete task: {response.status_code}", "details": response.text}
+        except Exception as e:
+            return {"success": False, "done": True, "error": "Failed to complete task.", "details": str(e)}
+
+@mcp.tool()
+async def delete_task(user_id: str, task_id: int) -> dict:
+    """
+    Remove a task.
+    """
+    if not SERVICE_SECRET:
+        return {"success": False, "done": True, "error": "Internal authentication error: Service secret missing."}
+    headers = _get_auth_headers()
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.delete(f"{BUSINESS_SERVICE_URL}/api/{user_id}/tasks/{task_id}", headers=headers)
+            if response.status_code in [200, 204]:
+                return {"success": True, "done": True, "message": "Task deleted successfully."}
+            else:
+                return {"success": False, "done": True, "error": f"Failed to delete task: {response.status_code}", "details": response.text}
+        except Exception as e:
+            return {"success": False, "done": True, "error": "Failed to delete task.", "details": str(e)}
+
+@mcp.tool()
+async def update_task(user_id: str, task_id: int, title: Optional[str] = None, description: Optional[str] = None, priority: Optional[str] = None) -> dict:
+    """
+    Modify task details.
+    """
+    if not SERVICE_SECRET:
+        return {"success": False, "done": True, "error": "Internal authentication error: Service secret missing."}
+    if title is None and description is None and priority is None:
+        return {"success": False, "done": True, "error": "At least one of 'title', 'description', or 'priority' must be provided."}
+    payload = {k: v for k, v in {"title": title, "description": description, "priority": priority}.items() if v is not None}
+    headers = {"Content-Type": "application/json", **_get_auth_headers()}
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.put(f"{BUSINESS_SERVICE_URL}/api/{user_id}/tasks/{task_id}", json=payload, headers=headers)
+            if response.status_code in [200, 204]:
+                return {"success": True, "done": True, "message": f"Task {task_id} updated successfully."}
+            else:
+                return {"success": False, "done": True, "error": f"Failed to update task: {response.status_code}", "details": response.text}
+        except Exception as e:
+            return {"success": False, "done": True, "error": "Failed to update task.", "details": str(e)}
+
+@mcp.tool()
+async def set_priority(user_id: str, task_id: int, priority: str) -> dict:
+    """
+    Set or update a task's priority.
+    """
+    if not SERVICE_SECRET:
+        return {"success": False, "done": True, "error": "Internal authentication error: Service secret missing."}
+    priority = priority.lower()
+    if priority not in ["low", "medium", "high"]:
+        return {"success": False, "done": True, "error": "Priority must be one of: 'low', 'medium', 'high'."}
+    payload = {"priority": priority}
+    headers = {"Content-Type": "application/json", **_get_auth_headers()}
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.put(f"{BUSINESS_SERVICE_URL}/api/{user_id}/tasks/{task_id}", json=payload, headers=headers)
+            if response.status_code in [200, 204]:
+                return {"success": True, "done": True, "message": f"Priority for task {task_id} updated successfully."}
+            else:
+                return {"success": False, "done": True, "error": f"Failed to update priority: {response.status_code}", "details": response.text}
+        except Exception as e:
+            return {"success": False, "done": True, "error": "Failed to update priority.", "details": str(e)}
+
+@mcp.tool()
+async def list_tasks_by_priority(user_id: str, priority: str, status: Literal["all", "pending", "completed"] = "all") -> dict:
+    """
+    Retrieve tasks filtered by priority.
+    """
+    if not SERVICE_SECRET:
+        return {"success": False, "done": True, "error": "Internal authentication error: Service secret missing."}
+    priority = priority.lower()
+    if priority not in ["low", "medium", "high"]:
+        return {"success": False, "done": True, "error": "Priority must be one of: 'low', 'medium', 'high'."}
+    params = {"priority": priority}
+    if status != "all":
+        params["status"] = status
+    headers = _get_auth_headers()
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.get(f"{BUSINESS_SERVICE_URL}/api/{user_id}/tasks", params=params, headers=headers)
+            if response.status_code == 200:
+                result = response.json()
+                tasks = result.get("tasks", [])
+                task_list = [{"id": t.get("id"), "title": t.get("title"), "priority": t.get("priority"), "completed": t.get("completed", False), "description": t.get("description"), "created_at": t.get("created_at")} for t in tasks]
+                return {"success": True, "done": True, "tasks": task_list, "count": len(task_list), "priority": priority, "status": status}
+            else:
+                return {"success": False, "done": True, "error": f"Failed to list tasks by priority: {response.status_code}", "details": response.text}
+        except Exception as e:
+            return {"success": False, "done": True, "error": "Failed to list tasks by priority.", "details": str(e)}
+
+if __name__ == "__main__":
+    try:
+        logger.info("MCP script running in __main__")
+        mcp.run()
+    except Exception as e:
+        logger.error("An unhandled exception occurred in ai_mcp_server.py", exc_info=True)
+        raise

config/config.py

@@ -0,0 +1,22 @@
+import os
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from pydantic import HttpUrl
+
+from dotenv import load_dotenv
+
+# Load environment variables from .env file
+load_dotenv()
+
+class Settings(BaseSettings):
+
+    GEMINI_API_KEY: str = os.getenv("GEMINI_API_KEY")
+    BUSINESS_SERVICE_URL: HttpUrl = "https://abdullahcoder54-todo-app.hf.space"
+    CONVERSATION_RETENTION_DAYS: int = 7
+    BETTER_AUTH_SECRET: str = os.getenv("BETTER_AUTH_SECRET")
+
+    model_config = {
+        "env_file": ".env",
+        "case_sensitive": True,
+        "extra": "allow"
+    }
+settings = Settings()

guardrails.py

@@ -0,0 +1,110 @@
+"""Guardrails for the TodoAgent."""
+
+from agents import (
+    Agent,
+    OpenAIChatCompletionsModel,
+    input_guardrail,
+    output_guardrail,
+    GuardrailFunctionOutput,
+    RunContextWrapper, # Needed for guardrail function signature
+    Runner, # Needed to run topic_checker_agent
+    TResponseInputItem # Correct type for input content in guardrail functions
+)
+from config.config import settings
+from openai import AsyncOpenAI
+import logging
+from typing import Any # Needed for RunContextWrapper[Any]
+from pydantic import BaseModel
+
+logger = logging.getLogger(__name__)
+
+class todo(BaseModel): 
+    reasoning: str
+    is_not_todo: bool
+    user_friendly_response: str
+
+# A simple, low-cost model to act as a topic checker.
+guardrail_client = AsyncOpenAI(
+    api_key=settings.GEMINI_API_KEY,
+    base_url="https://openrouter.ai/api/v1"
+)
+guardrail_model = OpenAIChatCompletionsModel(
+    model="arcee-ai/trinity-large-preview:free", # Using a common model for topic checking
+    openai_client=guardrail_client
+)
+
+# This is the "topic checker" agent.
+guardrail_agent = Agent(
+    name="TopicChecker",
+    model=guardrail_model,
+    output_type=todo,
+    instructions="""
+Your job is to check whether the user's message is related to managing a to-do list.
+
+If the message IS related to task management:
+- Adding a task
+- Deleting a task
+- Updating a task
+- Listing tasks
+- Completing tasks
+- Asking about existing tasks
+- Any clear todo/task management request
+
+
+If the message is NOT related to task management
+(e.g., math questions, weather, casual conversation, general knowledge):
+- Respond politely and briefly.
+- Explain that you can help only with managing tasks.
+- Guide the user to ask a task-related question.
+
+Your response MUST be friendly, clear, and short.
+
+Example response for non-task queries:
+"I’m here to help with your to-do list. You can ask me to add, remove, update, or list your tasks."
+
+"""
+)
+
+@input_guardrail(name="TodoInputGuardrail")
+async def is_task_related(
+    ctx: RunContextWrapper[Any], # Context provided by Runner
+    agent: Agent, # The agent running this guardrail
+    user_input: str | list[TResponseInputItem] # The user's input
+) -> GuardrailFunctionOutput:
+    """
+    Input guardrail to check if the user's query is related to task management.
+    """
+    try:
+
+        logger.info(f"Checking topic for query: '{user_input}'")
+
+        # Run the topic checker agent. Pass the input directly, and context for traceability.
+        result = await Runner.run(guardrail_agent, input=user_input, context=ctx.context)
+        logger.info(f"Topic checker result: is_todo={result.final_output.is_not_todo}, reasoning='{result.final_output.reasoning}'")
+        return GuardrailFunctionOutput(
+            output_info=result.final_output.user_friendly_response, 
+            tripwire_triggered=result.final_output.is_not_todo,
+        )
+
+    except Exception as e:
+        logger.error(f"Error in input guardrail: {e}")
+        return GuardrailFunctionOutput(
+            tripwire_triggered=False,
+            output_info="An error occurred while checking your query.",
+        )
+
+@output_guardrail(name="TodoOutputGuardrail")
+async def is_response_appropriate(
+    ctx: RunContextWrapper[Any], # Context provided by Runner
+    agent: Agent, # The agent running this guardrail
+    response_content: str | list[TResponseInputItem] # The agent's output
+) -> GuardrailFunctionOutput:
+    """
+    Output guardrail to check if the agent's response is appropriate.
+    (Placeholder - currently allows all responses.)
+    """
+    # In a real implementation, you might check for profanity, sensitive data, etc.
+    return GuardrailFunctionOutput(
+        tripwire_triggered=False,
+        output_info=None,
+    )

main.py

@@ -0,0 +1,267 @@
+"""FastAPI entrypoint for the ChatKit backend with TodoAgent."""
+
+from __future__ import annotations
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, Response, StreamingResponse
+from typing import Optional, Dict, Any
+import logging
+from contextlib import asynccontextmanager
+from uuid import uuid4
+
+from chatkit.server import ChatKitServer, StreamingResult
+from chatkit.store import Store, Page
+from chatkit.types import ThreadMetadata, UserMessageItem, ThreadItem, AssistantMessageItem, ThreadItemAddedEvent, ErrorEvent
+import json
+from todo_agent import TodoAgent
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class MemoryStore(Store):
+    """In-memory implementation of ChatKit Store for development."""
+
+    def __init__(self):
+        self.threads: Dict[str, ThreadMetadata] = {}
+        self.items: Dict[str, Dict[str, Any]] = {}
+        self.attachments: Dict[str, Any] = {}
+
+    async def load_threads(self, *, context: Dict[str, Any] | None = None, **kwargs) -> Page:
+        """Load all threads for a user, extracting user_id from context."""
+        user_id = context.get("user_id") if context else None
+        if not user_id:
+            # If no user_id is found, return an empty page.
+            return Page(data=[], has_more=False)
+            
+        threads = [t for t in self.threads.values() if t.user and t.user.id == user_id]
+        return Page(data=threads, has_more=False)
+
+    async def load_thread(self, thread_id: str, *args, context: Dict[str, Any] | None = None, **kwargs) -> ThreadMetadata | None:
+        """Load a single thread, create if doesn't exist."""
+        thread = self.threads.get(thread_id)
+        if thread is None:
+            # Create a new thread if it doesn't exist to prevent NoneType errors downstream.
+            user_id = context.get("user_id", "unknown") if context else "unknown"
+            thread = ThreadMetadata(
+                id=thread_id,
+                user_id=user_id,
+                created_at=0,  # Using a simple timestamp
+                updated_at=0,
+            )
+            self.threads[thread_id] = thread
+        return thread
+
+    async def save_thread(self, thread: ThreadMetadata, *args, context: Dict[str, Any] | None = None, **kwargs) -> None:
+        """Save a thread."""
+        self.threads[thread.id] = thread
+
+    async def delete_thread(self, thread_id: str, *args, context: Dict[str, Any] | None = None, **kwargs) -> None:
+        """Delete a thread."""
+        self.threads.pop(thread_id, None)
+        self.items.pop(thread_id, None)
+
+    async def load_thread_items(self, thread_id: str, *args, context: Dict[str, Any] | None = None, **kwargs) -> Page:
+        """Load all items for a thread."""
+        items = list(self.items.get(thread_id, {}).values())
+        return Page(data=items, has_more=False)
+
+    async def load_item(self, thread_id: str, item_id: str, *args, context: Dict[str, Any] | None = None, **kwargs) -> ThreadItem | None:
+        """Load a single item."""
+        return self.items.get(thread_id, {}).get(item_id)
+
+    async def save_item(self, thread_id: str, item: ThreadItem, *args, context: Dict[str, Any] | None = None, **kwargs) -> None:
+        """Save an item."""
+        if thread_id not in self.items:
+            self.items[thread_id] = {}
+        self.items[thread_id][item.id] = item
+
+    async def add_thread_item(self, thread_id: str, item: ThreadItem, *args, context: Dict[str, Any] | None = None, **kwargs) -> None:
+        """Add an item to a thread."""
+        await self.save_item(thread_id, item, context=context)
+
+    async def delete_thread_item(self, thread_id: str, item_id: str, *args, context: Dict[str, Any] | None = None, **kwargs) -> None:
+        """Delete an item from a thread."""
+        if thread_id in self.items:
+            self.items[thread_id].pop(item_id, None)
+
+    async def save_attachment(self, attachment: Any, *args, context: Dict[str, Any] | None = None, **kwargs) -> None:
+        """Save an attachment."""
+        self.attachments[attachment.id] = attachment
+
+    async def load_attachment(self, attachment_id: str, *args, context: Dict[str, Any] | None = None, **kwargs) -> Any | None:
+        """Load an attachment."""
+        return self.attachments.get(attachment_id)
+
+    async def delete_attachment(self, attachment_id: str, *args, context: Dict[str, Any] | None = None, **kwargs) -> None:
+        """Delete an attachment."""
+        self.attachments.pop(attachment_id, None)
+
+
+# Global instances
+todo_agent: Optional[TodoAgent] = None
+chatkit_server: Optional[ChatKitServer] = None
+
+
+class TodoChatKitServer(ChatKitServer):
+    """ChatKit server that wraps the TodoAgent."""
+
+    def __init__(self, todo_agent: TodoAgent):
+        self.todo_agent = todo_agent
+        super().__init__(MemoryStore())
+
+    async def respond(self, thread: ThreadMetadata, input_user_message: UserMessageItem | None, context: Dict[str, Any]):
+        """Implement the abstract respond method to handle user messages."""
+        try:
+            user_id = context.get("user_id", "unknown")
+            
+            # Get the message content
+            message_content = ""
+            if input_user_message and hasattr(input_user_message, 'content'):
+                message_content = input_user_message.content
+            
+            logger.info(f"Processing message from user {user_id}: {message_content}")
+            
+            # Use TodoAgent to generate response
+            response = await self.todo_agent.process_message(user_id, message_content)
+            
+            # Create an AssistantMessageItem for the response, ensuring all fields are correctly formatted.
+            assistant_item = AssistantMessageItem(
+                id=f"msg_{uuid4()}",
+                thread_id=thread.id,  # Pass the thread_id
+                created_at=0,  # Pass a timestamp
+                content=[{"type": "output_text", "text": response}],  # Format content as a list of blocks
+            )
+            
+            # Yield ThreadItemAddedEvent to add the assistant message to the thread
+            yield ThreadItemAddedEvent(
+                type="thread.item.added",
+                thread_id=thread.id,
+                item=assistant_item
+            )
+            
+        except Exception as e:
+            logger.error(f"Error in respond(): {e}", exc_info=True)
+            yield ErrorEvent(
+                type="error",
+                error=str(e)
+            )
+
+    async def process(self, payload: bytes | str, context: Dict[str, Any]) -> StreamingResult | dict | str:
+        """Process ChatKit events using TodoAgent."""
+        try:
+            # Parse the incoming payload
+            if isinstance(payload, bytes):
+                body = json.loads(payload.decode('utf-8'))
+            else:
+                body = json.loads(payload)
+
+            logger.info(f"Processing ChatKit event: {body.get('type')}")
+
+            # Extract user_id from context or payload
+            user_id = str(context.get("user_id", body.get("user", {}).get("id", "unknown")))
+
+            # Call the parent process method which handles ChatKit protocol
+            result = await super().process(payload, context)
+
+            return result
+
+        except Exception as e:
+            logger.error(f"Error in ChatKitServer.process(): {e}", exc_info=True)
+            return {"type": "error", "content": str(e)}
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Initialize agents on startup, cleanup on shutdown."""
+    global todo_agent, chatkit_server
+
+    logger.info("Initializing TodoAgent...")
+    todo_agent = TodoAgent()
+    logger.info("TodoAgent initialized")
+
+    logger.info("Initializing ChatKit Server...")
+    chatkit_server = TodoChatKitServer(todo_agent)
+    logger.info("ChatKit Server initialized")
+
+    yield
+
+    logger.info("Shutting down...")
+    todo_agent = None
+    chatkit_server = None
+
+
+app = FastAPI(
+    title="ChatKit Backend with TodoAgent",
+    description="ChatKit backend powered by TodoAgent",
+    version="1.0.0",
+    lifespan=lifespan
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.post("/chatkit")
+async def chatkit_endpoint(request: Request) -> Response:
+    """Proxy the ChatKit web component payload to the server implementation."""
+    global chatkit_server
+
+    if chatkit_server is None:
+        logger.error("ChatKit server not initialized")
+        return JSONResponse({"error": "Server not initialized"}, status_code=500)
+
+    try:
+        payload = await request.body()
+        
+        # Extract user_id from the request if available
+        context = {"request": request}
+        try:
+            import json
+            body = json.loads(payload.decode('utf-8'))
+            if "user" in body and "id" in body["user"]:
+                context["user_id"] = str(body["user"]["id"])
+        except Exception as e:
+            logger.debug(f"Could not extract user_id from payload: {e}")
+
+        result = await chatkit_server.process(payload, context)
+
+        if isinstance(result, StreamingResult):
+            return StreamingResponse(result, media_type="text/event-stream")
+        if hasattr(result, "json"):
+            return Response(content=result.json, media_type="application/json")
+        return JSONResponse(result)
+
+    except Exception as e:
+        logger.error(f"Error in chatkit_endpoint: {e}", exc_info=True)
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+
+@app.get("/")
+async def health_check():
+    """Health check endpoint."""
+    return {
+        "service": "chatkit-todo-assistant",
+        "initialized": chatkit_server is not None
+    }
+
+@app.get("/health")
+async def health():
+        return{"status": "ok"}
+
+@app.post("/chatkit/api")
+async def chatkit_api_endpoint(request: Request) -> Response:
+    """Alias for /chatkit endpoint for compatibility."""
+    return await chatkit_endpoint(request)
+
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

models/conversation.py

@@ -0,0 +1,33 @@
+from datetime import datetime
+from typing import List, Optional
+from pydantic import BaseModel, Field
+import uuid
+
+class Message(BaseModel):
+    role: str
+    content: str
+
+class Conversation(BaseModel):
+    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
+    user_id: str
+    messages: List[Message] = Field(default_factory=list)
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_at: datetime = Field(default_factory=datetime.utcnow)
+
+    def add_message(self, role: str, content: str):
+        self.messages.append(Message(role=role, content=content))
+        self.updated_at = datetime.utcnow()
+
+    def to_dict(self):
+        return {
+            "id": self.id,
+            "user_id": self.user_id,
+            "messages": [msg.model_dump() for msg in self.messages],
+            "created_at": self.created_at.isoformat(),
+            "updated_at": self.updated_at.isoformat(),
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict):
+        data["messages"] = [Message(**msg) for msg in data["messages"]]
+        return cls(**data)

pyproject.toml

@@ -0,0 +1,32 @@
+[project]
+name = "ai-service"
+version = "0.1.0"
+description = "AI Agent Service for Todo List App"
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+    "fastapi>=0.115.0",
+    "uvicorn[standard]>=0.32.0",
+    "openai-agents>=0.3.2",
+    "python-mcp>=1.0.0",
+    "httpx>=0.27.2",
+    "pydantic>=2.10",
+    "pydantic-settings>=2.6.1",
+    "openai-chatkit>=1.4.0"
+]
+
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools]
+py-modules = ["app", "todo_agent", "ai_mcp_server", "main"]
+
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["config*", "models*"]
+
+[tool.uv.workspace]
+members = [
+    "ai-service",
+]

requirements.txt

@@ -0,0 +1,7 @@
+fastapi==0.115.0
+uvicorn[standard]==0.32.0
+openai-agents==0.2.9
+python-mcp==1.0.0
+httpx==0.27.2
+pydantic==2.10
+pydantic-settings==2.6.1

todo_agent.py

@@ -0,0 +1,116 @@
+from typing import Dict, Any, List, Optional
+from agents import (
+    Agent,
+    Runner,
+    RunConfig,
+    OpenAIChatCompletionsModel,
+    ModelSettings,
+    set_tracing_disabled,
+    enable_verbose_stdout_logging,
+    InputGuardrailTripwireTriggered # Import the exception
+)
+from agents.mcp import MCPServerStdio
+from guardrails import is_task_related # Import the guardrail function
+import logging
+from openai import AsyncOpenAI
+import json
+import sys
+import os
+from config.config import settings
+
+set_tracing_disabled(disabled=True)
+enable_verbose_stdout_logging()
+logger = logging.getLogger(__name__)
+
+class TodoAgent:
+    def __init__(self):
+        self.client = AsyncOpenAI(
+            api_key=settings.GEMINI_API_KEY,
+            base_url="https://openrouter.ai/api/v1"
+        )
+        model = OpenAIChatCompletionsModel(
+            model="z-ai/glm-4.5-air:free",
+            openai_client=self.client
+        )
+        self.config = RunConfig(model=model, model_provider=self.client)
+        self.mcp_server = MCPServerStdio(
+            name="Todo Management MCP Server",
+            params={"command": sys.executable, "args": ["-m", "ai_mcp_server"]},
+            client_session_timeout_seconds=30.0
+        )
+        self.agent = Agent(
+            name="TodoAssistant",
+            instructions="""
+
+You are a Todo Management Assistant.
+
+YOUR RESPONSIBILITIES:
+- Add tasks when the user asks to create a todo.
+- List tasks when the user asks to see todos.
+- Update tasks when the user asks to modify a todo.
+- Delete tasks when the user asks to remove a todo.
+- Answer questions about existing tasks clearly.
+ 
+YOUR TOOLS:
+1. ADD_TASK - Adds a new task to the todo list.
+
+2. LIST_TASKS - Lists all current tasks in the todo list.
+
+3. UPDATE_TASK - Updates an existing task.
+
+4. DELETE_TASK - Deletes a task from the todo list.
+
+5. GET_TASK - Retrieves details of a specific task.
+
+CRITICAL EXECUTION RULES:
+
+- When a tool call succeeds, STOP immediately.
+- Do NOT repeat or retry the same tool call.
+- Do NOT perform multiple actions for a single request.
+- Your job ends after the correct tool is called and a brief confirmation is returned.
+
+
+TASK HANDLING RULES:
+1. A task consists of a clear title and optional details.
+2. Never create duplicate tasks with the same title.
+3. If a task already exists, inform the user instead of adding it again.
+4. When listing tasks, show them in a clear and readable format.
+5. When a task is added, updated, or deleted, confirm the action briefly.
+6. If the request is unclear, ask a short clarification question.
+7. Perform only the action the user requested — no extra actions.
+
+GENERAL BEHAVIOR:
+- Be concise and practical.
+- Do not invent tasks.
+- Do not repeat actions.
+- Always reflect the current state of the todo list.
+
+Your goal is to keep the user's todo list accurate and easy to manage.
+""",
+            mcp_servers=[self.mcp_server],
+            model_settings=ModelSettings(parallel_tool_calls=False),
+            input_guardrails=[is_task_related] # Add the input guardrail here
+        )
+
+    async def process_message(self, user_id: str, message: str) -> str:
+        """
+        Processes a user message and returns only the final string response,
+        as required by the simple, non-streaming backend template.
+        """
+        await self.mcp_server.connect()
+        
+        try:
+            result = await Runner.run(
+                self.agent,
+                input=f"[USER_ID: {user_id}] {message}",
+                run_config=self.config,
+
+            )
+            final_response = result.final_output if result.final_output else "I was able to process your request."
+
+        except InputGuardrailTripwireTriggered as e:
+            logger.warning(f"Input guardrail tripped for user {user_id}. Message: '{message}'")
+            final_response = e.guardrail_result.output.output_info
+        
+        logger.info(f"Agent for user {user_id} produced final response: {final_response}")
+        return  final_response