Oliver/presenton
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

update: simplify mcp

Browse source
This commit is contained in:
Suraj Jha 2025-08-17 23:16:26 +05:45
parent 292e945089
commit 3c9a4f9ce8
No known key found for this signature in database
GPG key ID: 5AC6C16355CE2C14
29 changed files with 288 additions and 1523 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
.cursor/mcp.json Normal file
View file

@ -0,0 +1,7 @@
{
"mcpServers": {
"presentation-generator": {
"url": "http://localhost:5000/mcp"
}
}
}

2
nginx.conf
View file

@ -31,7 +31,7 @@ http {
}
# MCP
location /mcp/ {
location /mcp {
proxy_pass http://localhost:8001;
proxy_read_timeout 30m;
proxy_connect_timeout 30m;

66
servers/fastapi/api/v1/ppt/endpoints/presentation.py
View file

@ -3,12 +3,12 @@ import json
import os
import random
from typing import Annotated, List, Literal, Optional
from fastapi import APIRouter, Body, Depends, File, HTTPException, UploadFile
from fastapi import APIRouter, Body, Depends, HTTPException
from fastapi.responses import StreamingResponse
from sqlalchemy import delete
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select
from constants.documents import UPLOAD_ACCEPTED_FILE_TYPES
from models.generate_presentation_request import GeneratePresentationRequest
from models.presentation_and_path import PresentationPathAndEditPath
from models.presentation_from_template import GetPresentationUsingTemplateRequest
from models.presentation_outline_model import (
@ -19,7 +19,7 @@ from models.pptx_models import PptxPresentationModel
from models.presentation_layout import PresentationLayoutModel
from models.presentation_structure_model import PresentationStructureModel
from models.presentation_with_slides import PresentationWithSlides
from services.score_based_chunker import ScoreBasedChunker
from utils.get_layout_by_name import get_layout_by_name
from services.icon_finder_service import IconFinderService
from services.image_generation_service import ImageGenerationService
@ -28,9 +28,9 @@ from utils.export_utils import export_presentation
from utils.llm_calls.generate_presentation_outlines import generate_ppt_outline
from models.sql.slide import SlideModel
from models.sse_response import SSECompleteResponse, SSEResponse
from services import TEMP_FILE_SERVICE
from services.database import get_async_session
from services.documents_loader import DocumentsLoader
from services import TEMP_FILE_SERVICE
from models.sql.presentation import PresentationModel
from services.pptx_presentation_creator import PptxPresentationCreator
from utils.asset_directory_utils import get_exports_directory, get_images_directory
@ -42,7 +42,7 @@ from utils.llm_calls.generate_slide_content import (
)
from utils.process_slides import process_slide_and_fetch_assets
from utils.randomizers import get_random_uuid
from utils.validators import validate_files
PRESENTATION_ROUTER = APIRouter(prefix="/presentation", tags=["Presentation"])
@ -310,53 +310,21 @@ async def create_pptx(
@PRESENTATION_ROUTER.post("/generate", response_model=PresentationPathAndEditPath)
async def generate_presentation_api(
prompt: Annotated[str, Body()],
n_slides: Annotated[int, Body()] = 8,
language: Annotated[str, Body()] = "English",
template: Annotated[str, Body()] = "general",
files: Annotated[Optional[List[UploadFile]], File()] = None,
export_as: Annotated[Literal["pptx", "pdf"], Body()] = "pptx",
request: GeneratePresentationRequest,
sql_session: AsyncSession = Depends(get_async_session),
):
validate_files(files, True, True, 50, UPLOAD_ACCEPTED_FILE_TYPES)
presentation_id = get_random_uuid()
temp_dir = TEMP_FILE_SERVICE.create_temp_dir()
# 1. Save uploaded files
file_paths = []
if files:
for upload in files:
file_path = os.path.join(temp_dir, upload.filename)
with open(file_path, "wb") as f:
f.write(await upload.read())
file_paths.append(file_path)
# 3. Generate Outlines
presentation_outlines = None
additional_context = ""
if file_paths:
documents_loader = DocumentsLoader(file_paths=file_paths)
await documents_loader.load_documents(temp_dir)
documents = documents_loader.documents
if documents:
additional_context = documents[0]
chunker = ScoreBasedChunker()
try:
chunks = await chunker.get_n_chunks(documents[0], n_slides)
presentation_outlines = PresentationOutlineModel(
slides=[chunk.to_slide_outline() for chunk in chunks]
)
except Exception as e:
print(e)
if not presentation_outlines:
presentation_outlines_text = ""
async for chunk in generate_ppt_outline(
prompt,
n_slides,
language,
request.prompt,
request.n_slides,
request.language,
additional_context,
):
presentation_outlines_text += chunk
@ -370,14 +338,14 @@ async def generate_presentation_api(
detail="Failed to generate presentation outlines. Please try again.",
)
presentation_outlines = PresentationOutlineModel(**presentation_outlines_json)
outlines = presentation_outlines.slides[:n_slides]
outlines = presentation_outlines.slides[:request.n_slides]
total_outlines = len(outlines)
print("-" * 40)
print(f"Generated {total_outlines} outlines for the presentation")
# 4. Parse Layouts
layout_model = await get_layout_by_name(template)
layout_model = await get_layout_by_name(request.template)
total_slide_layouts = len(layout_model.slides)
# 5. Generate Structure
@ -403,9 +371,9 @@ async def generate_presentation_api(
# 6. Create PresentationModel
presentation = PresentationModel(
id=presentation_id,
prompt=prompt,
n_slides=n_slides,
language=language,
prompt=request.prompt,
n_slides=request.n_slides,
language=request.language,
outlines=presentation_outlines.model_dump(),
layout=layout_model.model_dump(),
structure=presentation_structure.model_dump(),
@ -422,7 +390,7 @@ async def generate_presentation_api(
slide_layout = layout_model.slides[slide_layout_index]
print(f"Generating content for slide {i} with layout {slide_layout.id}")
slide_content = await get_slide_content_from_type_and_outline(
slide_layout, outlines[i], language
slide_layout, outlines[i], request.language
)
slide = SlideModel(
presentation=presentation_id,
@ -453,7 +421,7 @@ async def generate_presentation_api(
# 9. Export
presentation_and_path = await export_presentation(
presentation_id, presentation.title or get_random_uuid(), export_as
presentation_id, presentation.title or get_random_uuid(), request.export_as
)
return PresentationPathAndEditPath(

1
servers/fastapi/app_mcp/__init__.py
View file

@ -1 +0,0 @@
# This file marks the mcp directory as a Python package.

13
servers/fastapi/app_mcp/server.py
View file

@ -1,13 +0,0 @@
from fastmcp import FastMCP
from app_mcp.tools import register_tools
from app_mcp.services.workflow_orchestrator import WorkflowOrchestrator
def create_mcp_server():
mcp = FastMCP("PresentonMCP")
orchestrator = WorkflowOrchestrator()
register_tools(mcp, orchestrator)
return mcp
uvicorn_config = {
"reload": True,
}

0
servers/fastapi/app_mcp/services/__init__.py
View file

0
servers/fastapi/app_mcp/services/state_machine/__init__.py
View file

119
servers/fastapi/app_mcp/services/state_machine/constants.py
View file

@ -1,119 +0,0 @@
from app_mcp.services.state_machine.states import PresentationState
TRANSITIONS = {
PresentationState.INIT: {
PresentationState.OUTLINE_REQUESTED
},
# Outline generation flow (now includes file processing)
PresentationState.OUTLINE_REQUESTED: {
PresentationState.OUTLINE_GENERATED,
PresentationState.OUTLINE_FAILED
},
PresentationState.OUTLINE_GENERATED: {
PresentationState.OUTLINE_APPROVED,
PresentationState.OUTLINE_REQUESTED,
PresentationState.OUTLINE_FAILED
},
PresentationState.OUTLINE_APPROVED: {
PresentationState.LAYOUT_REQUESTED
},
# Layout selection flow
PresentationState.LAYOUT_REQUESTED: {
PresentationState.LAYOUT_SELECTED
},
PresentationState.LAYOUT_SELECTED: {
PresentationState.GENERATION_IN_PROGRESS,
PresentationState.LAYOUT_REQUESTED
},
# Presentation generation flow
PresentationState.GENERATION_IN_PROGRESS: {
PresentationState.PRESENTATION_READY,
PresentationState.GENERATION_FAILED
},
PresentationState.PRESENTATION_READY: {
PresentationState.EXPORT_REQUESTED,
PresentationState.EDIT_REQUESTED,
PresentationState.OUTLINE_REQUESTED
},
# Export flow
PresentationState.EXPORT_REQUESTED: {
PresentationState.EXPORT_IN_PROGRESS
},
PresentationState.EXPORT_IN_PROGRESS: {
PresentationState.EXPORT_COMPLETE,
PresentationState.EXPORT_FAILED
},
PresentationState.EXPORT_COMPLETE: {
PresentationState.EDIT_REQUESTED,
PresentationState.EXPORT_REQUESTED,
PresentationState.INIT
},
# Edit and revision flow
PresentationState.EDIT_REQUESTED: {
PresentationState.TEMPLATE_EDITING
},
PresentationState.TEMPLATE_EDITING: {
PresentationState.PRESENTATION_READY,
PresentationState.EDIT_FAILED
},
# Error recovery transitions
PresentationState.OUTLINE_FAILED: {
PresentationState.OUTLINE_REQUESTED,
PresentationState.INIT
},
PresentationState.GENERATION_FAILED: {
PresentationState.LAYOUT_SELECTED,
PresentationState.OUTLINE_APPROVED
},
PresentationState.EXPORT_FAILED: {
PresentationState.EXPORT_REQUESTED,
PresentationState.PRESENTATION_READY
},
PresentationState.EDIT_FAILED: {
PresentationState.EDIT_REQUESTED,
PresentationState.PRESENTATION_READY
}
}
SUGGESTIONS = {
PresentationState.INIT: "Start with outline generation (files will be processed automatically if provided)",
PresentationState.OUTLINE_REQUESTED: "Generating presentation outline with file analysis if applicable",
PresentationState.OUTLINE_GENERATED: "Review and approve outline",
PresentationState.OUTLINE_APPROVED: "Select presentation layout",
PresentationState.LAYOUT_SELECTED: "Generate presentation",
PresentationState.PRESENTATION_READY: "Export presentation or request edits",
PresentationState.EXPORT_REQUESTED: "Choose export format and generate",
PresentationState.EXPORT_COMPLETE: "Download presentation or start new one",
PresentationState.EDIT_REQUESTED: "Make template-based edits",
}
PROGRESS_WEIGHTS = {
PresentationState.INIT: 0,
PresentationState.OUTLINE_REQUESTED: 20,
PresentationState.OUTLINE_GENERATED: 35,
PresentationState.OUTLINE_APPROVED: 40,
PresentationState.LAYOUT_REQUESTED: 45,
PresentationState.LAYOUT_SELECTED: 50,
PresentationState.GENERATION_IN_PROGRESS: 70,
PresentationState.PRESENTATION_READY: 85,
PresentationState.EXPORT_REQUESTED: 90,
PresentationState.EXPORT_IN_PROGRESS: 95,
PresentationState.EXPORT_COMPLETE: 100,
PresentationState.TEMPLATE_EDITING: 60,
}
ERROR_STATES = {
PresentationState.OUTLINE_FAILED,
PresentationState.GENERATION_FAILED,
PresentationState.EXPORT_FAILED,
PresentationState.EDIT_FAILED
}

19
servers/fastapi/app_mcp/services/state_machine/context.py
View file

@ -1,19 +0,0 @@
from typing import Dict, Set, Optional, Any
from dataclasses import dataclass
@dataclass
class StateContext:
"""Context data that travels with the state machine"""
presentation_id: Optional[str] = None
title: Optional[str] = None
outlines: Optional[list] = None
layout: Optional[str] = None
file_paths: Optional[list] = None
export_format: Optional[str] = None
export_path: Optional[str] = None
error_message: Optional[str] = None
metadata: Dict[str, Any] = None
def __post_init__(self):
if self.metadata is None:
self.metadata = {}

101
servers/fastapi/app_mcp/services/state_machine/machine.py
View file

@ -1,101 +0,0 @@
from typing import Dict, Set, Any
from app_mcp.services.state_machine.context import StateContext
from app_mcp.services.state_machine.states import PresentationState
from app_mcp.services.state_machine.constants import TRANSITIONS, SUGGESTIONS, PROGRESS_WEIGHTS, ERROR_STATES
class PresentationStateMachine:
def __init__(self):
self.state = PresentationState.INIT
self.context = StateContext()
self._state_history = [PresentationState.INIT]
self._transitions = TRANSITIONS
self._error_states = ERROR_STATES
self._suggestions = SUGGESTIONS
self._progress_weights = PROGRESS_WEIGHTS
def transition(self, new_state: PresentationState, context_updates: Dict[str, Any] = None):
"""
Transition to new state with optional context updates
Args:
new_state (PresentationState): The state to transition to
context_updates (Dict[str, Any], optional): Context data to update during transition
Raises:
ValueError: If the transition is not valid
"""
if not self.is_valid_transition(new_state):
raise ValueError(f"Invalid transition from {self.state} to {new_state}")
# Update context if provided
if context_updates:
for key, value in context_updates.items():
if hasattr(self.context, key):
setattr(self.context, key, value)
else:
self.context.metadata[key] = value
# Record state history
self._state_history.append(new_state)
self.state = new_state
def is_valid_transition(self, new_state: PresentationState) -> bool:
"""Check if transition to new state is valid"""
return new_state in self._transitions.get(self.state, set())
def get_available_transitions(self) -> Set[PresentationState]:
"""Get all valid transitions from current state"""
return self._transitions.get(self.state, set())
def can_transition_to(self, target_state: PresentationState) -> bool:
"""Check if can transition to target state"""
return target_state in self.get_available_transitions()
def is_terminal_state(self) -> bool:
"""Check if current state is terminal (no outgoing transitions)"""
return len(self.get_available_transitions()) == 0
def is_error_state(self) -> bool:
"""Check if current state is an error state"""
return self.state in self._error_states
def get_workflow_progress(self) -> float:
"""Calculate workflow progress as percentage"""
return self._progress_weights.get(self.state, 0)
def get_next_suggested_action(self) -> str:
"""Get suggested next action based on current state"""
return self._suggestions.get(self.state, "No suggestions available")
def reset(self):
"""Reset state machine to initial state"""
self.state = PresentationState.INIT
self.context = StateContext()
self._state_history = [PresentationState.INIT]
def get_state_history(self) -> list:
"""Get history of states visited"""
return self._state_history.copy()
def rollback_to_previous_state(self) -> bool:
"""Rollback to previous state if possible"""
if len(self._state_history) < 2:
return False
# Remove current state from history
self._state_history.pop()
previous_state = self._state_history[-1]
if self.is_valid_transition(previous_state):
self.state = previous_state
return True
else:
self._state_history.append(self.state)
return False
def __str__(self):
return f"PresentationStateMachine(state={self.state.name}, progress={self.get_workflow_progress()}%)"
def __repr__(self):
return (f"PresentationStateMachine(state={self.state.name}, "
f"context={self.context}, "
f"history_length={len(self._state_history)})")

35
servers/fastapi/app_mcp/services/state_machine/states.py
View file

@ -1,35 +0,0 @@
from enum import Enum, auto
class PresentationState(Enum):
"""
Represents the various states in the presentation workflow.
"""
INIT = auto()
# Outline generation phase (now includes file processing)
OUTLINE_REQUESTED = auto()
OUTLINE_GENERATED = auto()
OUTLINE_APPROVED = auto()
# Layout selection phase
LAYOUT_REQUESTED = auto()
LAYOUT_SELECTED = auto()
# Presentation generation phase
GENERATION_IN_PROGRESS = auto()
PRESENTATION_READY = auto()
# Export phase
EXPORT_REQUESTED = auto()
EXPORT_IN_PROGRESS = auto()
EXPORT_COMPLETE = auto()
# Edit and revision loops
EDIT_REQUESTED = auto()
TEMPLATE_EDITING = auto()
# Error states
OUTLINE_FAILED = auto()
GENERATION_FAILED = auto()
EXPORT_FAILED = auto()
EDIT_FAILED = auto()

308
servers/fastapi/app_mcp/services/workflow_orchestrator.py
View file

@ -1,308 +0,0 @@
from typing import Dict, Any, Optional, List
from dataclasses import asdict
from app_mcp.services.state_machine.machine import PresentationStateMachine
from app_mcp.services.state_machine.states import PresentationState
from utils.user_config import update_env_with_user_config
from app_mcp.wrapper.generate_outline import generate_outline
from app_mcp.wrapper.presentation_generation import process_post_outline_workflow
from app_mcp.wrapper.presentation_export import export_presentation_and_get_path
from app_mcp.wrapper.list_layout import list_layouts
class WorkflowOrchestrator:
"""
Orchestrates the presentation generation workflow using FSM
- Handles session management
- Executes
- file uploads
- summary generation
- outline generation
- layout selection
- presentation generation
- export
- Provides status and context management
- Allows for session-based operations
- Supports error handling and recovery
"""
def __init__(self):
"""
Initiating:
- The environment with user configuration from the user config file.
- The Finite State Machine (FSM) for presentation workflow.
- Active sessions dictionary to manage multiple workflows.
"""
try:
update_env_with_user_config()
except Exception as e:
print(f"Error updating environment with user config: {e}")
self.fsm = PresentationStateMachine()
self._active_sessions: Dict[str, PresentationStateMachine] = {}
def create_session(self, session_id: str) -> PresentationStateMachine:
"""
Create a new workflow session with the given session ID.
If a session with the same ID already exists, it will be replaced.
Session will Remain for the lifetime of the application.
Args:
session_id (str): Unique identifier for the session.
"""
if not session_id or not isinstance(session_id, str):
raise ValueError("Session ID must be a non-empty string")
session_id = session_id.strip()
if not session_id:
raise ValueError("Session ID cannot be empty")
if session_id in self._active_sessions:
self.remove_session(session_id)
print(f"Session {session_id} already exists, replacing it.")
self._active_sessions[session_id] = PresentationStateMachine()
return self._active_sessions[session_id]
def get_session(self, session_id: str) -> Optional[PresentationStateMachine]:
"""Get existing workflow session"""
if not session_id or not isinstance(session_id, str):
return None
return self._active_sessions.get(session_id.strip())
def remove_session(self, session_id: str) -> bool:
"""Remove workflow session"""
return self._active_sessions.pop(session_id, None) is not None
async def execute_generate_outline(self, session_id: str, prompt: str, **kwargs) -> Dict[str, Any]:
"""
Execute outline generation workflow step
Args:
session_id (str): Unique identifier for the session.
prompt (str): The prompt to generate the outline.
**kwargs: Additional parameters for outline generation.
Returns:
Dict[str, Any]: Result containing status, state, progress, next action, and generated outline.
"""
fsm = self.get_session(session_id)
if not fsm:
raise ValueError(f"Session {session_id} not found")
try:
fsm.transition(PresentationState.OUTLINE_REQUESTED)
result = await generate_outline(prompt, **kwargs)
# Update the Context and transition to outline generated
context_updates = {
"title": result["title"],
"outlines": result["outlines"]
}
fsm.transition(PresentationState.OUTLINE_GENERATED, context_updates)
return {
"status": "success",
"state": fsm.state.name,
"progress": fsm.get_workflow_progress(),
"next_action": "Review outline and approve",
"result": result,
"can_approve": True
}
except Exception as e:
fsm.transition(PresentationState.OUTLINE_FAILED, {"error_message": str(e)})
print(f"Error generating outline for session {session_id}: {e}")
return {
"status": "error",
"state": fsm.state.name,
"error": str(e),
"next_action": fsm.get_next_suggested_action()
}
async def approve_outline(self, session_id: str) -> Dict[str, Any]:
"""
Approve the generated outline
Args:
session_id (str): Unique identifier for the session.
Returns:
Dict[str, Any]: Result containing status, state, progress, next action.
"""
fsm = self.get_session(session_id)
if not fsm:
raise ValueError(f"Session {session_id} not found")
if fsm.state != PresentationState.OUTLINE_GENERATED:
raise ValueError(f"Cannot approve outline in state {fsm.state.name}")
fsm.transition(PresentationState.OUTLINE_APPROVED)
return {
"status": "success",
"state": fsm.state.name,
"progress": fsm.get_workflow_progress(),
"next_action": fsm.get_next_suggested_action()
}
async def execute_layout_selection(self, session_id: str, layout: str) -> Dict[str, Any]:
"""
Execute layout selection workflow step
Args:
session_id (str): Unique identifier for the session.
layout (str): Selected layout for the presentation.
Returns:
Dict[str, Any]: Result containing status, state, progress, next action, and selected layout.
"""
fsm = self.get_session(session_id)
if not fsm:
raise ValueError(f"Session {session_id} not found")
try:
fsm.transition(PresentationState.LAYOUT_REQUESTED)
#Updating the context and transitioning to LAYOUT_SELECTED
context_updates = {"layout": layout}
fsm.transition(PresentationState.LAYOUT_SELECTED, context_updates)
return {
"status": "success",
"state": fsm.state.name,
"progress": fsm.get_workflow_progress(),
"next_action": fsm.get_next_suggested_action(),
"selected_layout": layout
}
except Exception as e:
print(f"Error selecting layout for session {session_id}: {e}")
return {
"status": "error",
"error": str(e),
"next_action": "Please select a valid layout"
}
async def execute_presentation_generation(self, session_id: str, **kwargs) -> Dict[str, Any]:
"""
Execute presentation generation workflow step
Args:
session_id (str): Unique identifier for the session.
**kwargs: Additional parameters for presentation generation.
Returns:
Dict[str, Any]: Result containing status, state, progress, next action, and generated presentation.
"""
fsm = self.get_session(session_id)
if not fsm:
raise ValueError(f"Session {session_id} not found")
try:
fsm.transition(PresentationState.GENERATION_IN_PROGRESS)
notes = kwargs.get('notes', [])
result = await process_post_outline_workflow(
title=fsm.context.title,
outlines=fsm.context.outlines,
notes=notes,
layout=fsm.context.layout,
prompt=fsm.context.metadata.get('original_prompt', ""),
sql_session=None,
**kwargs
)
#Updating the Context and transitioning to PRESENTATION_READY
context_updates = {"presentation_id": result["presentation_id"]}
fsm.transition(PresentationState.PRESENTATION_READY, context_updates)
return {
"status": "success",
"state": fsm.state.name,
"progress": fsm.get_workflow_progress(),
"next_action": fsm.get_next_suggested_action(),
"result": result
}
except Exception as e:
fsm.transition(PresentationState.GENERATION_FAILED, {"error_message": str(e)})
print(f"Error generating presentation for session {session_id}: {e}")
return {
"status": "error",
"state": fsm.state.name,
"error": str(e),
"next_action": fsm.get_next_suggested_action()
}
async def execute_export(self, session_id: str, export_format: str = "pptx") -> Dict[str, Any]:
"""
Execute presentation export workflow step
Args:
session_id (str): Unique identifier for the session.
export_format (str): Format to export the presentation (e.g., "pptx", "pdf").
Returns:
Dict[str, Any]: Result containing status, state, progress, next action, and export
"""
fsm = self.get_session(session_id)
if not fsm:
raise ValueError(f"Session {session_id} not found")
try:
# Transition to EXPORT_REQUESTED state
fsm.transition(PresentationState.EXPORT_REQUESTED, {"export_format": export_format})
fsm.transition(PresentationState.EXPORT_IN_PROGRESS)
result = await export_presentation_and_get_path(
presentation_id=fsm.context.presentation_id,
title=fsm.context.title,
export_as=export_format
)
print("RResult of export:", result)
#Updating the Context and transitioning to EXPORT_COMPLETE
context_updates = {"export_path": result["path"]}
fsm.transition(PresentationState.EXPORT_COMPLETE, context_updates)
return {
"status": "success",
"state": fsm.state.name,
"progress": fsm.get_workflow_progress(),
"next_action": "Download your presentation or start a new one",
"result": result
}
except Exception as e:
fsm.transition(PresentationState.EXPORT_FAILED, {"error_message": str(e)})
print(f"Error exporting presentation for session {session_id}: {e}")
return {
"status": "error",
"state": fsm.state.name,
"error": str(e),
"next_action": fsm.get_next_suggested_action()
}
async def get_available_layouts(self) -> List[Any]:
"""
Get available presentation layouts
"""
return await list_layouts()
def get_workflow_status(self, session_id: str) -> Dict[str, Any]:
"""Get current workflow status"""
fsm = self.get_session(session_id)
if not fsm:
return {"error": "Session not found"}
return {
"session_id": session_id,
"current_state": fsm.state.name,
"progress": fsm.get_workflow_progress(),
"next_action": fsm.get_next_suggested_action(),
"available_transitions": [s.name for s in fsm.get_available_transitions()],
"is_error_state": fsm.is_error_state(),
"context": asdict(fsm.context),
"state_history": [s.name for s in fsm.get_state_history()]
}
def get_all_sessions(self) -> Dict[str, Dict[str, Any]]:
"""
Get status of all active sessions
"""
return {
session_id: self.get_workflow_status(session_id)
for session_id in self._active_sessions.keys()
}

35
servers/fastapi/app_mcp/tools/__init__.py
View file

@ -1,35 +0,0 @@
"""MCP Tools package for presentation generation."""
from app_mcp.tools.choose_layout import register_choose_layout
from app_mcp.tools.export_presentation import register_export_presentation
from app_mcp.tools.get_status import register_get_status
from app_mcp.tools.show_layouts import register_show_layouts
from app_mcp.tools.start_presentation import register_start_presentation
from app_mcp.tools.help_me import register_help_me
from app_mcp.tools.continue_workflow import register_continue_workflow
__all__ = [
'register_choose_layout',
'register_export_presentation',
'register_get_status',
'register_show_layouts',
'register_start_presentation',
'register_help_me',
'register_continue_workflow',
'register_tools',
]
def register_tools(mcp, orchestrator):
"""Register all MCP tools in a fancy way."""
tools = [
register_choose_layout,
register_export_presentation,
register_get_status,
register_show_layouts,
register_start_presentation,
register_help_me,
register_continue_workflow
]
for tool in tools:
tool(mcp, orchestrator)

46
servers/fastapi/app_mcp/tools/choose_layout.py
View file

@ -1,46 +0,0 @@
from typing import Dict, Any
def register_choose_layout(mcp, orchestrator):
"""Register all workflow-related tools for chat-based interaction"""
@mcp.tool("choose_layout")
async def choose_layout(session_id: str, layout_name: str) -> Dict[str, Any]:
"""
🎨 Select a visual style and theme for your presentation.
Choose from available professional layouts that determine:
- Color scheme and visual design
- Slide structure and layout patterns
- Font choices and styling
- Overall presentation aesthetic
Use 'show_layouts' first to see all available options. Only show the layout name and short description.
Args:
session_id: Your presentation session ID
layout_name: Name of the layout you want to use
"""
try:
result = await orchestrator.execute_layout_selection(session_id, layout_name)
if result["status"] == "success":
return {
"status": "success",
"session_id": session_id,
"message": f"Perfect! I've selected the '{layout_name}' layout for your presentation.",
"suggestion": "Now I'll generate all the slides with content, images, and styling. This might take a minute or two.",
"available_actions": {
"continue": "Start generating the presentation",
"change_layout": "Actually, let me pick a different layout"
}
}
return result
except Exception as e:
return {
"status": "error",
"error": str(e),
"session_id": session_id
}
return choose_layout

129
servers/fastapi/app_mcp/tools/continue_workflow.py
View file

@ -1,129 +0,0 @@
from typing import Dict, Any
def register_continue_workflow(mcp, orchestrator):
"""Register all workflow-related tools for chat-based interaction"""
@mcp.tool("continue_workflow")
async def continue_workflow(
session_id: str,
action: str = "continue"
) -> Dict[str, Any]:
"""
⏭️ Move to the next step in creating your presentation.
This tool automatically determines what should happen next based on where
you are in the process:
- After starting: Generates your presentation outline
- After outline: Shows available layouts to choose from
- After layout: Creates your complete presentation
Just call this when you're ready to proceed to the next step!
Args:
session_id: Your presentation session ID
action: What to do next (usually just "continue")
"""
try:
# Validate session_id
if not session_id or not isinstance(session_id, str):
return {
"status": "error",
"error": "Valid session_id is required",
"suggestion": "Use the same session_id from start_presentation"
}
session_id = session_id.strip()
fsm = orchestrator.get_session(session_id)
if not fsm:
return {
"status": "error",
"error": "Session not found. Please start a new presentation first.",
"suggestion": "Call start_presentation to begin"
}
current_state = fsm.state.name
if current_state in ["INIT"]:
# Generate outline (this now handles file processing internally)
prompt = fsm.context.metadata.get("original_prompt", "")
n_slides = fsm.context.metadata.get("n_slides", 8)
language = fsm.context.metadata.get("language", "English")
files = fsm.context.metadata.get("files", None)
if not prompt:
return {
"status": "error",
"error": "No prompt found in session. Please start over.",
"suggestion": "Call start_presentation with a valid prompt"
}
# Pass files to outline generation if they exist
kwargs = {"n_slides": n_slides, "language": language}
if files:
kwargs["files"] = files
result = await orchestrator.execute_generate_outline(
session_id, prompt, **kwargs
)
if result["status"] == "success":
return {
"status": "success",
"session_id": session_id,
"message": "Here's your presentation outline:",
"title": result["result"]["title"],
"outlines": result["result"]["outlines"],
"files_processed": bool(files),
"suggestion": "Take a look at the outline. If it looks good, use 'continue_workflow' again to proceed to layout selection.",
"next_step": "Call continue_workflow again to choose layouts"
}
return result
elif current_state == "OUTLINE_GENERATED":
# Auto-approve and move to layouts
await orchestrator.approve_outline(session_id)
layouts = await orchestrator.get_available_layouts()
return {
"status": "success",
"session_id": session_id,
"message": "Great! Now let's choose a visual style for your presentation.",
"available_layouts": layouts,
"suggestion": "Choose a layout that fits your content and audience. Use 'choose_layout' with the layout name.",
"next_step": "Call choose_layout with your preferred layout name"
}
elif current_state == "LAYOUT_SELECTED":
# Generate presentation
result = await orchestrator.execute_presentation_generation(session_id)
if result["status"] == "success":
return {
"status": "success",
"session_id": session_id,
"message": "🎉 Your presentation is ready!",
"title": result["result"]["title"],
"presentation_id": result["result"]["presentation_id"],
"suggestion": "Your presentation has been generated successfully! Use 'export_presentation' to download it.",
"next_step": "Call export_presentation with format 'pptx' or 'pdf'"
}
return result
else:
return {
"status": "info",
"message": f"Currently in {current_state} state.",
"suggestion": "Use get_status to see what actions are available.",
"next_step": "Call get_status for guidance"
}
except Exception as e:
return {
"status": "error",
"error": f"Workflow error: {str(e)}",
"session_id": session_id if 'session_id' in locals() else "unknown",
"suggestion": "Use get_status to check your current progress"
}
return continue_workflow

51
servers/fastapi/app_mcp/tools/export_presentation.py
View file

@ -1,51 +0,0 @@
from typing import Dict, Any
def register_export_presentation(mcp, orchestrator):
"""Register all workflow-related tools for chat-based interaction"""
@mcp.tool("export_presentation")
async def export_presentation(
session_id: str,
format: str = "pptx",
export_path: str = None
) -> Dict[str, Any]:
"""
📁 Download your finished presentation in your preferred format.
Export your completed presentation as:
- "pptx" - PowerPoint format (editable, best for sharing and presenting)
- "pdf" - PDF format (read-only, best for viewing and printing)
The exported file will be ready for download immediately.
Args:
session_id: Your presentation session ID
format: Export format - either "pptx" or "pdf"
"""
try:
if format.lower() not in ["pdf", "pptx"]:
return {
"status": "error",
"error": "Please choose either 'pdf' or 'pptx' format",
"session_id": session_id
}
result = await orchestrator.execute_export(session_id, format.lower())
print("Export result:", result)
if result["status"] == "success":
return {
"status": "success",
"session_id": session_id,
"message": f"🎉 Your presentation has been exported as {format.upper()}!",
"path": result["result"]["path"],
"suggestion": "You can download it now, or start creating another presentation."
}
return result
except Exception as e:
return {
"status": "error",
"error": str(e),
"session_id": session_id
}
return export_presentation

83
servers/fastapi/app_mcp/tools/get_status.py
View file

@ -1,83 +0,0 @@
from typing import Dict, Any, Optional, List
def register_get_status(mcp, orchestrator):
"""Register all workflow-related tools for chat-based interaction"""
@mcp.tool("get_status")
def get_status(session_id: str) -> Dict[str, Any]:
"""
📊 Check your presentation creation progress.
See exactly where you are in the process:
- What step you're currently on
- How much progress you've made
- What you can do next
- Any issues that need attention
Perfect for checking in if you're unsure what to do next!
Args:
session_id: Your presentation session ID
"""
try:
if not session_id or not isinstance(session_id, str):
return {
"status": "error",
"error": "Valid session_id is required"
}
session_id = session_id.strip()
status = orchestrator.get_workflow_status(session_id)
if "error" in status:
return {
"status": "error",
"error": "Session not found. Start a new presentation with 'start_presentation'.",
"available_sessions": list(orchestrator._active_sessions.keys())
}
state = status["current_state"]
# Provide user-friendly status messages
friendly_messages = {
"INIT": "Ready to start! Use 'start_presentation' to begin.",
"OUTLINE_REQUESTED": "Generating outline with file analysis if applicable.",
"OUTLINE_GENERATED": "Outline created. Use 'continue_workflow' to proceed to layouts.",
"OUTLINE_APPROVED": "Outline approved. Use 'choose_layout' to select a theme.",
"LAYOUT_SELECTED": "Layout chosen. Use 'continue_workflow' to generate presentation.",
"PRESENTATION_READY": "Presentation generated! Use 'export_presentation' to download.",
"EXPORT_COMPLETE": "All done! Presentation exported successfully."
}
next_actions = {
"INIT": "start_presentation",
"OUTLINE_REQUESTED": "Wait for outline generation to complete",
"OUTLINE_GENERATED": "continue_workflow",
"OUTLINE_APPROVED": "choose_layout",
"LAYOUT_SELECTED": "continue_workflow",
"PRESENTATION_READY": "export_presentation",
"EXPORT_COMPLETE": "Download file or start_presentation for new one"
}
return {
"status": "success",
"session_id": session_id,
"current_step": state,
"progress": f"{status['progress']:.0f}%",
"message": friendly_messages.get(state, f"Currently in {state} state"),
"next_action": next_actions.get(state, status["next_action"]),
"context": {
"prompt": status["context"].get("metadata", {}).get("original_prompt"),
"n_slides": status["context"].get("metadata", {}).get("n_slides"),
"language": status["context"].get("metadata", {}).get("language")
}
}
except Exception as e:
return {
"status": "error",
"error": f"Status check failed: {str(e)}",
"suggestion": "Try start_presentation to begin a new session"
}
return get_status

48
servers/fastapi/app_mcp/tools/help_me.py
View file

@ -1,48 +0,0 @@
from typing import Dict, Any, Optional, List
def register_help_me(mcp, orchestrator):
"""Register all workflow-related tools for chat-based interaction"""
@mcp.tool("help")
def help() -> Dict[str, Any]:
"""
❓ Get help and guidance for creating presentations.
Shows you:
- Step-by-step workflow guide
- Available commands and what they do
- Example usage to get you started
- Tips for best results
Perfect for first-time users or when you need a refresher!
"""
return {
"status": "info",
"message": "🎯 Complete Guide to Creating Presentations",
"workflow": {
"step_1": "🚀 start_presentation - Begin with your topic and optional files",
"step_2": "📋 continue_workflow - Generate and review your outline",
"step_3": "🎨 choose_layout - Pick a visual style that fits your content",
"step_4": "⚡ continue_workflow - Generate your complete presentation",
"step_5": "📁 export_presentation - Download as PowerPoint or PDF"
},
"helpful_commands": {
"get_status": "📊 Check your current progress anytime",
"show_layouts": "👀 Browse available themes and styles",
"help": "❓ Show this helpful guide"
},
"quick_start": {
"with_files": "start_presentation(session_id='my-session', prompt='Your topic', files=[uploaded_files])",
"text_only": "start_presentation(session_id='my-session', prompt='Create a presentation about sustainable energy')",
"custom": "start_presentation(session_id='my-session', prompt='Your topic', n_slides=10, language='Spanish')"
},
"tips": [
"💡 Be specific in your prompt for better results",
"📎 Upload relevant files to enhance your content",
"🎨 Choose layouts that match your audience and purpose",
"📊 Use get_status anytime to see what's next"
]
}
return help

39
servers/fastapi/app_mcp/tools/show_layouts.py
View file

@ -1,39 +0,0 @@
from typing import Dict, Any
def register_show_layouts(mcp, orchestrator):
"""Register all workflow-related tools for chat-based interaction"""
@mcp.tool("show_layouts")
async def show_layouts(session_id: str) -> Dict[str, Any]:
"""
👀 Browse all available presentation themes and layouts.
See the complete list of professional layouts including:
- Business and corporate themes
- Creative and modern designs
- Academic and educational styles
- Technical and data-focused layouts
Each layout comes with its own color scheme, fonts, and slide structures.
Args:
session_id: Your presentation session ID
"""
try:
layouts = await orchestrator.get_available_layouts()
return {
"status": "success",
"session_id": session_id,
"message": "Here are all the available presentation layouts:",
"layouts": layouts,
"suggestion": "Choose one using 'choose_layout' with the layout name."
}
except Exception as e:
return {
"status": "error",
"error": str(e),
"session_id": session_id
}
return show_layouts

111
servers/fastapi/app_mcp/tools/start_presentation.py
View file

@ -1,111 +0,0 @@
from typing import List, Dict, Any, Optional
def register_start_presentation(mcp, orchestrator):
"""Register all workflow-related tools for chat-based interaction"""
@mcp.tool("start_presentation")
async def start_presentation(
session_id: str,
prompt: str,
files: Optional[List] = None,
n_slides: int = 8,
language: str = "English"
) -> Dict[str, Any]:
"""
🚀 Start creating a new presentation with your idea!
This is your entry point to create presentations. You can:
- Start with just a text prompt describing what you want
- Upload files (PDFs, docs, etc.) to base your presentation on
- Specify how many slides you want (default: 8)
- Choose the language for your presentation
Examples:
- "Create a presentation about climate change solutions"
- "Make slides about our Q4 financial results" (with uploaded files)
- "Build a training deck for new employees"
Args:
session_id: Unique identifier for your presentation session
prompt: Describe what your presentation should be about
files: Optional list of files to analyze and include
n_slides: Number of slides to generate (default: 8)
language: Presentation language (default: English)
"""
try:
if not session_id or not isinstance(session_id, str) or len(session_id.strip()) == 0:
return {
"status": "error",
"error": "Session ID is required and must be a non-empty string",
"example": "Use something like: session_id='my_presentation_123'"
}
if not prompt or not isinstance(prompt, str) or len(prompt.strip()) == 0:
return {
"status": "error",
"error": "Prompt is required and must be a non-empty string",
"example": "prompt='Create a presentation about AI in healthcare'"
}
# Clean session_id
session_id = session_id.strip()
# Create session
orchestrator.create_session(session_id)
# Store initial parameters
fsm = orchestrator.get_session(session_id)
if not fsm:
return {
"status": "error",
"error": "Failed to create session",
"session_id": session_id
}
fsm.context.metadata.update({
"original_prompt": prompt.strip(),
"n_slides": max(1, min(50, n_slides)), # Validate slide count
"language": language.strip() if language else "English"
})
# Debug log to verify metadata update
print("DEBUG: Metadata after update:", fsm.context.metadata)
# Handle files if provided - store them in context for later use
if files and len(files) > 0:
# Store files in context for integrated processing during outline generation
fsm.context.metadata.update({
"files": files
})
return {
"status": "success",
"session_id": session_id,
"message": "Great! I've received your files and will analyze them during presentation creation.",
"prompt": prompt,
"files_count": len(files),
"suggestion": f"Now I'll create a presentation outline based on your prompt '{prompt}' and analyze the uploaded files. Use 'continue_workflow' to proceed.",
"next_step": "Call continue_workflow to generate the outline with file analysis"
}
else:
# Direct outline generation without files
return {
"status": "success",
"session_id": session_id,
"message": f"Perfect! Let's create a presentation about: '{prompt}'",
"suggestion": "I'll generate an outline with the key topics and structure. Use 'continue_workflow' to proceed.",
"next_step": "Call continue_workflow to generate the outline",
"parameters": {
"n_slides": fsm.context.metadata.get("n_slides", 8), # Ensure n_slides is retrieved correctly
"language": fsm.context.metadata.get("language", "English") # Ensure language is retrieved correctly
}
}
except Exception as e:
return {
"status": "error",
"error": f"Unexpected error: {str(e)}",
"session_id": session_id if 'session_id' in locals() else "unknown",
"suggestion": "Please try again with a valid session_id and prompt"
}
return start_presentation

52
servers/fastapi/app_mcp/wrapper/edit_from_template.py
View file

@ -1,52 +0,0 @@
from typing import Dict, Any
from models.sql.presentation import PresentationModel
from models.sql.slide import SlideModel
from models.presentation_from_template import GetPresentationUsingTemplateRequest
from utils.dict_utils import deep_update
from utils.export_utils import export_presentation
from sqlmodel import select
from fastapi import HTTPException
class EditFromTemplateTools:
def __init__(self):
pass
def register(self, mcp):
@mcp.tool("edit_from_template")
async def edit_from_template(
data: GetPresentationUsingTemplateRequest,
sql_session
) -> Dict[str, Any]:
"""
Create a new presentation from a template and updated slide data, then export.
"""
presentation = await sql_session.get(PresentationModel, data.presentation_id)
if not presentation:
raise HTTPException(status_code=404, detail="Presentation not found")
slides = await sql_session.scalars(
select(SlideModel).where(SlideModel.presentation == data.presentation_id)
)
new_presentation = presentation.get_new_presentation()
new_slides = []
for each_slide in slides:
updated_content = None
new_slide_data = list(filter(lambda x: x.index == each_slide.index, data.data))
if new_slide_data:
updated_content = deep_update(each_slide.content, new_slide_data[0].content)
new_slides.append(
each_slide.get_new_slide(new_presentation.id, updated_content)
)
sql_session.add(new_presentation)
sql_session.add_all(new_slides)
await sql_session.commit()
presentation_and_path = await export_presentation(
new_presentation.id, new_presentation.title, data.export_as
)
return {
**presentation_and_path.model_dump(),
"edit_path": f"/presentation?id={new_presentation.id}",
}

97
servers/fastapi/app_mcp/wrapper/generate_outline.py
View file

@ -1,97 +0,0 @@
import json
import os
from fastapi import HTTPException
from typing import Dict, Any, Optional, List, Annotated
from models.presentation_outline_model import PresentationOutlineModel
from utils.llm_calls.generate_presentation_outlines import generate_ppt_outline
from services import TEMP_FILE_SERVICE
from services.documents_loader import DocumentsLoader
from services.score_based_chunker import ScoreBasedChunker
from utils.validators import validate_files
from fastapi import UploadFile, File
from constants.documents import UPLOAD_ACCEPTED_FILE_TYPES
import asyncio
async def generate_outline(
prompt: str,
n_slides: int = 8,
language: str = "English",
files: Annotated[Optional[List[UploadFile]], File()] = None,
) -> Dict[str, Any]:
"""
Generate presentation outlines given a prompt, number of slides, language, optional summary, and files.
Files are now processed directly within this function instead of a separate step.
Returns the parsed outline data.
"""
validate_files(files, True, True, 50, UPLOAD_ACCEPTED_FILE_TYPES)
temp_dir = TEMP_FILE_SERVICE.create_temp_dir()
file_paths = []
if files:
for upload in files:
file_path = os.path.join(temp_dir, upload.filename)
with open(file_path, "wb") as f:
f.write(await upload.read())
file_paths.append(file_path)
presentation_outlines = None
additional_context = ""
if file_paths:
documents_loader = DocumentsLoader(file_paths=file_paths)
await documents_loader.load_documents(temp_dir)
documents = documents_loader.documents
if documents:
additional_context = documents[0]
chunker = ScoreBasedChunker()
try:
chunks = await chunker.get_n_chunks(documents[0], n_slides)
presentation_outlines = PresentationOutlineModel(
slides=[chunk.to_slide_outline() for chunk in chunks]
)
except Exception as e:
print(e)
if not presentation_outlines:
presentation_outlines_text = ""
async for chunk in generate_ppt_outline(
prompt,
n_slides,
language,
additional_context,
):
# Give control to the event loop
await asyncio.sleep(0)
presentation_outlines_text += chunk
try:
presentation_outlines_json = json.loads(presentation_outlines_text)
presentation_outlines = PresentationOutlineModel(
**presentation_outlines_json
)
except Exception as e:
print(e)
raise HTTPException(
status_code=400,
detail="Failed to generate presentation outlines. Please try again.",
)
# Truncate slides to n_slides
presentation_outlines.slides = presentation_outlines.slides[:n_slides]
# Compose title from first slide
title = (
presentation_outlines.slides[0][:50]
.replace("#", "")
.replace("/", "")
.replace("\\", "")
.replace("\n", "")
)
# Prepare outlines list
outlines = presentation_outlines.model_dump(mode="json")
return {
"title": title,
"outlines": outlines,
}

8
servers/fastapi/app_mcp/wrapper/list_layout.py
View file

@ -1,8 +0,0 @@
from typing import List, Any
from api.v1.ppt.endpoints.layouts import get_layouts
async def list_layouts() -> List[Any]:
"""
Retrieve and return a list of all available presentation layouts.
"""
return await get_layouts()

25
servers/fastapi/app_mcp/wrapper/presentation_export.py
View file

@ -1,25 +0,0 @@
from typing import Literal, Dict, Any
from utils.export_utils import export_presentation
# Standalone function for workflow orchestrator
async def export_presentation_and_get_path(
presentation_id: str,
title: str,
export_as: Literal["pptx", "pdf"] = "pptx"
) -> Dict[str, Any]:
"""
Export the presentation and return the export path and edit path.
"""
presentation_and_path = await export_presentation(
presentation_id, title, export_as
)
# model_dump() is assumed to return a dict with the export path and related info
data = presentation_and_path.model_dump()
print("Exported presentation data:", data)
# Map export_path to path if needed
return {
**data,
"edit_path": f"/presentation?id={presentation_id}",
"export_path": data["path"],
}

126
servers/fastapi/app_mcp/wrapper/presentation_generation.py
View file

@ -1,126 +0,0 @@
import random
from typing import List, Dict, Any, Optional
from models.presentation_layout import PresentationLayoutModel
from models.presentation_structure_model import PresentationStructureModel
from models.sql.presentation import PresentationModel
from models.sql.slide import SlideModel
from utils.get_layout_by_name import get_layout_by_name
from utils.llm_calls.generate_presentation_structure import (
generate_presentation_structure,
)
from utils.llm_calls.generate_slide_content import (
get_slide_content_from_type_and_outline,
)
from services.image_generation_service import ImageGenerationService
from services.icon_finder_service import IconFinderService
from utils.asset_directory_utils import get_images_directory
from utils.process_slides import process_slide_and_fetch_assets
from models.presentation_outline_model import PresentationOutlineModel
from utils.randomizers import get_random_uuid
import asyncio
from sqlalchemy.ext.asyncio import AsyncSession
# Standalone function for workflow orchestrator
async def process_post_outline_workflow(
title: str,
outlines: List[str],
layout: str = "general",
language: str = "English",
prompt: str = "",
n_slides: int = 8,
sql_session: Optional[AsyncSession] = None,
) -> Dict[str, Any]:
"""
Process the workflow after outlines are generated: layout, structure, slides, assets, save, and ask for export.
"""
# 1. Parse Layout
layout_model: PresentationLayoutModel = await get_layout_by_name(layout)
total_slide_layouts = len(layout_model.slides)
# 2. Generate Structure
if layout_model.ordered:
presentation_structure = layout_model.to_presentation_structure()
else:
presentation_structure: PresentationStructureModel = (
await generate_presentation_structure(
presentation_outline=PresentationOutlineModel(
slides=outlines,
),
presentation_layout=layout_model,
)
)
presentation_structure.slides = presentation_structure.slides[:n_slides]
for index in range(n_slides):
random_slide_index = random.randint(0, total_slide_layouts - 1)
if index >= n_slides:
presentation_structure.slides.append(random_slide_index)
continue
if presentation_structure.slides[index] >= total_slide_layouts:
presentation_structure.slides[index] = random_slide_index
# 3. Create PresentationModel
presentation_id = get_random_uuid()
presentation = PresentationModel(
id=presentation_id,
title=title,
n_slides=n_slides,
language=language,
outlines=outlines,
prompt=prompt,
layout=layout_model.model_dump(),
structure=presentation_structure.model_dump(),
)
image_generation_service = ImageGenerationService(get_images_directory())
icon_finder_service = IconFinderService()
async_asset_generation_tasks = []
# 4. Generate slide content and save slides
slides: List[SlideModel] = []
for i, slide_layout_index in enumerate(presentation_structure.slides):
slide_layout = layout_model.slides[slide_layout_index]
slide_content = await get_slide_content_from_type_and_outline(
slide_layout, outlines[i], language
)
slide = SlideModel(
presentation=presentation_id,
layout_group=layout_model.name,
layout=slide_layout.id,
index=i,
content=slide_content,
)
async_asset_generation_tasks.append(
process_slide_and_fetch_assets(
image_generation_service, icon_finder_service, slide
)
)
slides.append(slide)
generated_assets_lists = await asyncio.gather(*async_asset_generation_tasks)
generated_assets = []
for assets_list in generated_assets_lists:
generated_assets.extend(assets_list)
# 5. Save PresentationModel and Slides
if sql_session is None:
from services.database import get_async_session
async for session in get_async_session():
session.add(presentation)
session.add_all(slides)
session.add_all(generated_assets)
await session.commit()
else:
sql_session.add(presentation)
sql_session.add_all(slides)
sql_session.add_all(generated_assets)
await sql_session.commit()
# 6. Ask user if they want to export and in which format
return {
"presentation_id": presentation_id,
"title": title,
"message": "Presentation is ready. Would you like to export? (pdf or pptx)",
"export_options": ["pdf", "pptx"],
}

87
servers/fastapi/mcp_server.py
View file

@ -1,24 +1,81 @@
import sys
import os
import argparse
import asyncio
import traceback
from urllib.parse import urljoin
import time
import httpx
from fastmcp import FastMCP
import json
from app_mcp.server import create_mcp_server, uvicorn_config
async def main():
parser = argparse.ArgumentParser(description="Run the FastAPI server")
parser.add_argument(
"--port", type=int, default=8001, help="Port number to run the server on"
)
args = parser.parse_args()
try:
print("DEBUG: MCP (OpenAPI) Server startup initiated")
parser = argparse.ArgumentParser(description="Run the MCP server (from OpenAPI)")
parser.add_argument(
"--port", type=int, default=8001, help="Port for the MCP HTTP server"
)
parser.add_argument(
"--api-base-url",
type=str,
default="http://127.0.0.1:8000",
help="Base URL of the FastAPI server to wrap (e.g., http://127.0.0.1:8000)",
)
parser.add_argument(
"--openapi-path",
type=str,
default="/openapi.json",
help="Path to the OpenAPI JSON on the FastAPI server",
)
parser.add_argument(
"--name",
type=str,
default="Presenton API (OpenAPI)",
help="Display name for the generated MCP server",
)
args = parser.parse_args()
print(
f"DEBUG: Parsed args - port={args.port}, api_base_url={args.api_base_url}, openapi_path={args.openapi_path}"
)
with open("openai_spec.json", "r") as f:
openapi_spec = json.load(f)
# Create an HTTP client that the MCP server will use to call the API
api_client = httpx.AsyncClient(base_url=args.api_base_url, timeout=60.0)
# Build MCP server from OpenAPI
print("DEBUG: Creating FastMCP server from OpenAPI spec...")
mcp = FastMCP.from_openapi(
openapi_spec=openapi_spec,
client=api_client,
name=args.name,
)
print("DEBUG: MCP server created from OpenAPI successfully")
# Start the MCP server
uvicorn_config = {"reload": True}
print(f"DEBUG: Starting MCP server on host=0.0.0.0, port={args.port}")
await mcp.run_async(
transport="http",
host="0.0.0.0",
port=args.port,
uvicorn_config=uvicorn_config,
)
print("DEBUG: MCP server run_async completed")
except Exception as e:
print(f"ERROR: MCP server startup failed: {e}")
print(f"ERROR: Traceback: {traceback.format_exc()}")
raise
mcp = create_mcp_server()
await mcp.run_async(
transport="http",
host="0.0.0.0",
port=args.port,
uvicorn_config=uvicorn_config
)
if __name__ == "__main__":
asyncio.run(main())
print("DEBUG: Starting MCP (OpenAPI) main function")
try:
asyncio.run(main())
except Exception as e:
print(f"FATAL ERROR: {e}")
print(f"FATAL TRACEBACK: {traceback.format_exc()}")
sys.exit(1)

10
servers/fastapi/models/generate_presentation_request.py Normal file
View file

@ -0,0 +1,10 @@
from typing import Literal, Optional
from pydantic import BaseModel, Field
class GeneratePresentationRequest(BaseModel):
prompt: str = Field(..., description="The prompt for generating the presentation")
n_slides: int = Field(default=8, description="Number of slides to generate")
language: str = Field(default="English", description="Language for the presentation")
template: str = Field(default="general", description="Template to use for the presentation")
export_as: Literal["pptx", "pdf"] = Field(default="pptx", description="Export format")

160
servers/fastapi/openai_spec.json Normal file
View file

@ -0,0 +1,160 @@
{
"openapi": "3.1.0",
"info": {
"title": "FastAPI",
"version": "0.1.0"
},
"paths": {
"/api/v1/ppt/presentation/generate": {
"post": {
"tags": ["Presentation"],
"summary": "Generate Presenatation MCP",
"operationId": "presentation_generator",
"requestBody": {
"content": {
"multipart/form-data": {
"schema": {
"$ref": "#/components/schemas/Body_generate_presentation_api_api_v1_ppt_presentation_generate_post"
}
}
},
"required": true
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/PresentationPathAndEditPath"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
}
},
"components": {
"schemas": {
"Body_generate_presentation_api_api_v1_ppt_presentation_generate_post": {
"properties": {
"prompt": {
"type": "string",
"title": "Prompt"
},
"n_slides": {
"type": "integer",
"title": "N Slides",
"default": 8
},
"language": {
"type": "string",
"title": "Language",
"default": "English"
},
"template": {
"type": "string",
"title": "Template",
"default": "general"
},
"files": {
"anyOf": [
{
"items": {
"type": "string",
"format": "binary"
},
"type": "array"
},
{
"type": "null"
}
],
"title": "Files"
},
"export_as": {
"type": "string",
"enum": ["pptx", "pdf"],
"title": "Export As",
"default": "pptx"
}
},
"type": "object",
"required": ["prompt"],
"title": "Body_generate_presentation_api_api_v1_ppt_presentation_generate_post"
},
"PresentationPathAndEditPath": {
"properties": {
"presentation_id": {
"type": "string",
"title": "Presentation Id"
},
"path": {
"type": "string",
"title": "Path"
},
"edit_path": {
"type": "string",
"title": "Edit Path"
}
},
"type": "object",
"required": ["presentation_id", "path", "edit_path"],
"title": "PresentationPathAndEditPath"
},
"HTTPValidationError": {
"properties": {
"detail": {
"items": {
"$ref": "#/components/schemas/ValidationError"
},
"type": "array",
"title": "Detail"
}
},
"type": "object",
"title": "HTTPValidationError"
},
"ValidationError": {
"properties": {
"loc": {
"items": {
"anyOf": [
{
"type": "string"
},
{
"type": "integer"
}
]
},
"type": "array",
"title": "Location"
},
"msg": {
"type": "string",
"title": "Message"
},
"type": {
"type": "string",
"title": "Error Type"
}
},
"type": "object",
"required": ["loc", "msg", "type"],
"title": "ValidationError"
}
}
}
}

33
start.js
View file

@ -104,19 +104,28 @@ const startServers = async () => {
console.error("FastAPI process failed to start:", err);
});
// const appmcpProcess = spawn(
// "python",
// ["mcp_server.py", "--port", appmcpPort.toString()],
// {
// cwd: fastapiDir,
// stdio: "inherit",
// env: process.env,
// },
// );
const appmcpProcess = spawn(
"python",
[
"mcp_server.py",
"--port",
appmcpPort.toString(),
"--api-base-url",
`http://127.0.0.1:${fastapiPort}`,
"--openapi-path",
"/openapi.json",
],
{
cwd: fastapiDir,
stdio: "inherit",
env: process.env,
},
);
appmcpProcess.on("error", (err) => {
console.error("App MCP process failed to start:", err);
});
// appmcpProcess.on("error", (err) => {
// console.error("App MCP process failed to start:", err);
// });
const nextjsProcess = spawn(
"npm",