FastMCP

"""FastMCP Echo Server.""" from enum import Enum from pathlib import Path from typing import Dict, List, Literal, Optional, Annotated from mcp import GetPromptResult from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp.exceptions import ValidationError from mcp.server.fastmcp.prompts.prompt_manager import Prompt from mcp.server.fastmcp.prompts.base import PromptArgument from mcp.types import TextContent, PromptMessage from pydantic import BaseModel, Field, EmailStr # Create server mcp = FastMCP("EchoPromptServer") # PROMPTS SETUP class MessageTaskInput(BaseModel): """ Input validation for echo task prompt. Attributes: name (str): Name of the user email (EmailStr): Email address of the user subscribe (bool): Whether the user wants to subscribe """ name: str = Field(..., min_length=1, max_length=100, description="Name of the user (str)") number: int = Field(...,description="your lucky number (int)") email: EmailStr = Field(..., description="Email address of the user (EmailStr)") subscribe: bool = Field(..., description="Whether the user wants to subscribe (bool)") EMAIL_TEMPLATE_SUBSCRIBE = """ Hello {name}, thanks for subscribing today. We'll send updates to {email}. {number} is your lucky number """ EMAIL_TEMPLATE_UNSUBSCRIBE = """ Hello {name}, you unsubscribed today. We will no longer send updates to {email}. {number} is your lucky number """ @mcp.prompt("basic") def basic_prompt( name: Annotated[str, Field(description="Name of the user (str)")], number: Annotated[int, Field(description="Lucky number (int)")], email: Annotated[EmailStr, Field(description="Contact email address (EmailStr|str)")], subscribe: Annotated[bool, Field(description="Whether to subscribe to updates (bool)")] ) -> GetPromptResult: """ Generate a text response using the provided input. Args: name (str): Name of the user. number (int): Lucky number. email (EmailStr): Email address of the user. subscribe (bool): Whether the user wants to subscribe. Returns: str: a message for the user. """ # Validate input using MessageTaskInput try: input_data = MessageTaskInput(name=name, number=number, email=email, subscribe=subscribe) if input_data.subscribe: prompt = EMAIL_TEMPLATE_SUBSCRIBE.format( name=input_data.name, email=input_data.email, number=input_data.number ) else: prompt = EMAIL_TEMPLATE_UNSUBSCRIBE.format( name=input_data.name, email=input_data.email, number=input_data.number ) prompt = prompt.strip() return GetPromptResult( description="py-code task execution template", messages=[ PromptMessage( role="user", content=TextContent(type="text", text=prompt), ), ], ) except ValidationError as e: return f"Validation error {e}" @mcp.prompt("basic_plus") def basic_plus_prompt( name: Annotated[ str, Field( description=MessageTaskInput.model_fields["name"].description, ), ], number: Annotated[ int, Field( description=MessageTaskInput.model_fields["number"].description, ), ], email: Annotated[ EmailStr, Field( description=MessageTaskInput.model_fields["email"].description, ), ], subscribe: Annotated[ bool, Field( description=MessageTaskInput.model_fields["subscribe"].description, ), ], ) -> GetPromptResult: """ A function that uses individual arguments, directly referencing the model for validation metadata. """ # Validate input using MessageTaskInput try: input_data = MessageTaskInput(name=name, number=number, email=email, subscribe=subscribe) if input_data.subscribe: prompt = EMAIL_TEMPLATE_SUBSCRIBE.format( name=input_data.name, email=input_data.email, number=input_data.number ) else: prompt = EMAIL_TEMPLATE_UNSUBSCRIBE.format( name=input_data.name, email=input_data.email, number=input_data.number ) prompt = prompt.strip() return GetPromptResult( description="py-code task execution template", messages=[ PromptMessage( role="user", content=TextContent(type="text", text=prompt), ), ], ) except ValidationError as e: return f"Validation error {e}" @mcp.prompt("typed") def typed_prompt( name: Annotated[str, Field(description=MessageTaskInput.model_fields["name"].description)], number: Annotated[int, Field(description=MessageTaskInput.model_fields["number"].description)], email: Annotated[EmailStr, Field(description=MessageTaskInput.model_fields["email"].description)], subscribe: Annotated[bool, Field(description=MessageTaskInput.model_fields["subscribe"].description)] ) -> GetPromptResult: """ Generate a text response using the provided input. Args: name (str): Name of the user number (int): Lucky number email (EmailStr): Email address of the user subscribe (bool): Whether to subscribe Returns: GetPromptResult: A result containing the formatted message for the user. """ input_data = MessageTaskInput(name=name, number=number, email=email, subscribe=subscribe) if input_data.subscribe: prompt = EMAIL_TEMPLATE_SUBSCRIBE.format( name=input_data.name, email=input_data.email, number=input_data.number ) else: prompt = EMAIL_TEMPLATE_UNSUBSCRIBE.format( name=input_data.name, email=input_data.email, number=input_data.number ) prompt = prompt.strip() return GetPromptResult( description="py-code task execution template", messages=[ PromptMessage( role="user", content=TextContent(type="text", text=prompt), ), ], ) class ProjectStatus(BaseModel): """Represents the current status and priority of a project, along with assigned users. Attributes: task (str): The name of the task. status (str): The current status of the project (e.g., "In Progress", "Completed"). priority (int): A numeric priority level (e.g., 1 for high priority). assignees (List[str]): A list of usernames assigned to this project status. """ task: str = Field(..., min_length=1, max_length=100, description="Name of the task (str)") status: str = Field(..., min_length=1, max_length=100, description="Status of the task (str)") priority: int = Field(..., description="Priority of the task (int)") assignees: List[str] = Field(..., description="list of assignees (List[str])") @mcp.prompt("mixed") def mixed_prompt( task: Annotated[str, Field(description=ProjectStatus.model_fields["task"].description)], status: Annotated[str, Field(description=ProjectStatus.model_fields["status"].description)], priority: Annotated[int, Field(description=ProjectStatus.model_fields["priority"].description)], assignees: Annotated[List[str], Field(description=ProjectStatus.model_fields["assignees"].description)] ) -> GetPromptResult: """ Generate a status report for a project. Args: task (str): Name of the task status (str): Current project status priority (int): Priority level assignees (List[str]): List of assigned usernames Returns: GetPromptResult: A result containing the formatted status report. """ input_data = ProjectStatus( task=task, status=status, priority=priority, assignees=assignees ) # Format the status report assignee_list = ", ".join(input_data.assignees) report = f""" Project Status Report -------------------- Task Name: {input_data.task} Status: {input_data.status} Priority Level: {input_data.priority} Assigned Team Members: {assignee_list} """ report = report.strip() return GetPromptResult( description="Project status report", messages=[ PromptMessage( role="user", content=TextContent(type="text", text=report), ), ], ) class UserProfile(BaseModel): """Represents a user profile with essential identity details. Attributes: username (str): The unique identifier for the user. email (str): The email address associated with the user. is_active (bool): Indicates whether the user's account is currently active. """ username: str = Field(..., min_length=1, max_length=50, description="The unique identifier for the user (str)") email: str = Field(..., min_length=5, max_length=100, description="The email address associated with the user (str)") is_active: bool = Field(..., description="Indicates whether the user's account is currently active (bool)") class ComplexInput(BaseModel): """Defines a complex input structure containing project details, ownership, status updates, and settings. Attributes: project_name (str): The name of the project. owner (UserProfile): The user profile of the project owner. statuses (List[ProjectStatus]): A list of status updates, including priority and assignees. settings (Dict[str, List[str]]): A dictionary of configuration settings, where keys represent setting categories and values are lists of associated options. """ project_name: str = Field(..., min_length=1, max_length=100, description="The name of the project (str)") owner: UserProfile = Field(..., description="The user profile of the project owner (UserProfile)") statuses: List[ProjectStatus] = Field(..., description="List of status updates, including priority and assignees (List[ProjectStatus])") settings: Dict[str, List[str]] = Field(..., description="Dictionary of configuration settings with categories and options (Dict[str, List[str]])") @mcp.prompt("complex") def complex_prompt( project_name: Annotated[str, Field(description=ComplexInput.model_fields["project_name"].description)], owner_username: Annotated[str, Field(description=UserProfile.model_fields["username"].description)], owner_email: Annotated[str, Field(description=UserProfile.model_fields["email"].description)], owner_is_active: Annotated[bool, Field(description=UserProfile.model_fields["is_active"].description)], statuses_task: Annotated[str, Field(description=ProjectStatus.model_fields["task"].description)], statuses_status: Annotated[str, Field(description=ProjectStatus.model_fields["status"].description)], statuses_priority: Annotated[int, Field(description=ProjectStatus.model_fields["priority"].description)], statuses_assignees: Annotated[ List[str], Field(description=ProjectStatus.model_fields["assignees"].description)], settings: Annotated[Dict[str, List[str]], Field(description=ComplexInput.model_fields["settings"].description)] ) -> GetPromptResult: """ Generate a comprehensive project report. Args: project_name (str): The name of the project owner_username (str): The unique identifier for the project owner owner_email (str): The email address of the project owner owner_is_active (bool): Whether the project owner's account is active statuses_task (str): Name of the task statuses_status (str): Current project status statuses_priority (int): Priority level statuses_assignees (List[str]): List of assigned usernames settings (Dict[str, List[str]]): Project configuration settings Returns: GetPromptResult: A result containing the formatted project report. """ # Construct nested objects owner = UserProfile( username=owner_username, email=owner_email, is_active=owner_is_active ) status = ProjectStatus( task=statuses_task, status=statuses_status, priority=statuses_priority, assignees=statuses_assignees ) input_data = ComplexInput( project_name=project_name, owner=owner, statuses=[status], # Note: Using single status for now settings=settings ) # Format the comprehensive report assignee_list = ", ".join(status.assignees) settings_str = "\n".join(f" {category}: {', '.join(values)}" for category, values in input_data.settings.items()) report = f""" Comprehensive Project Report -------------------------- Project Name: {input_data.project_name} Owner Information: - Username: {input_data.owner.username} - Email: {input_data.owner.email} - Account Status: {"Active" if input_data.owner.is_active else "Inactive"} Current Status: - Task: {status.task} - Status: {status.status} - Priority Level: {status.priority} - Assigned Team Members: {assignee_list} Project Settings: {settings_str} """ report = report.strip() return GetPromptResult( description="Comprehensive project report", messages=[ PromptMessage( role="user", content=TextContent(type="text", text=report), ), ], )