initial mcp server setup

This commit is contained in:
OwusuBlessing
2025-09-11 23:13:58 +01:00
commit 20f96c0f30
141 changed files with 14444 additions and 0 deletions
@@ -0,0 +1,7 @@
"""
Server Prompts Module
"""
from .base_prompt import BaseServerPrompt
from .prompt_registry import ServerPromptRegistry
__all__ = ['BaseServerPrompt', 'ServerPromptRegistry']
@@ -0,0 +1,51 @@
"""
Base Server Prompt Class
"""
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
class BaseServerPrompt(ABC):
"""Base class for server prompts that can be registered with FastMCP"""
def __init__(self, name: str, description: str, template: str, arguments: Optional[Dict[str, Any]] = None):
self.name = name
self.description = description
self.template = template
self.arguments = arguments or {}
@abstractmethod
async def generate(self, **kwargs) -> str:
"""Generate the prompt with the provided arguments"""
pass
def get_prompt_definition(self) -> Dict[str, Any]:
"""Get the prompt definition for FastMCP registration"""
return {
"name": self.name,
"description": self.description,
"arguments": self.arguments
}
def create_fastmcp_prompt(self, mcp_server):
"""Create a FastMCP prompt decorator for this prompt"""
@mcp_server.prompt()
async def prompt_wrapper(**kwargs):
return await self.generate(**kwargs)
# Set metadata
prompt_wrapper.__name__ = self.name
prompt_wrapper.__doc__ = self.description
return prompt_wrapper
def _substitute_template(self, **kwargs) -> str:
"""Helper method to substitute variables in template"""
result = self.template
# Apply provided arguments
for key, value in kwargs.items():
placeholder = f"{{{key}}}"
result = result.replace(placeholder, str(value))
return result
@@ -0,0 +1,44 @@
"""
Greeting Prompt Example
"""
from .base_prompt import BaseServerPrompt
class GreetingPrompt(BaseServerPrompt):
"""A greeting prompt template"""
def __init__(self):
super().__init__(
name="greeting_prompt",
description="Generate a greeting prompt for AI models",
template="Please write a {style} greeting for someone named {name}. The greeting should be {tone} and include a {element}.",
arguments={
"style": {
"type": "string",
"enum": ["friendly", "formal", "casual"],
"description": "Style of greeting",
"default": "friendly"
},
"tone": {
"type": "string",
"enum": ["warm", "professional", "relaxed"],
"description": "Tone of the greeting",
"default": "warm"
},
"element": {
"type": "string",
"enum": ["compliment", "question", "observation"],
"description": "Element to include in greeting",
"default": "compliment"
}
}
)
async def generate(self, name: str, style: str = "friendly", tone: str = "warm", element: str = "compliment") -> str:
"""Generate the greeting prompt"""
return self._substitute_template(
name=name,
style=style,
tone=tone,
element=element
)
@@ -0,0 +1,100 @@
"""
Server Prompt Registry for dynamic prompt registration
"""
import os
import importlib
import inspect
from typing import List, Dict, Any, Type, Optional
from pathlib import Path
from .base_prompt import BaseServerPrompt
class ServerPromptRegistry:
"""Registry for managing server prompts from files"""
def __init__(self, prompts_directory: str = None):
self.prompts_directory = prompts_directory or os.path.dirname(__file__)
self._registered_prompts: Dict[str, BaseServerPrompt] = {}
self._prompt_files: List[str] = []
@property
def directory(self):
"""Alias for prompts_directory for backward compatibility"""
return self.prompts_directory
@property
def _prompts(self):
"""Alias for _registered_prompts for backward compatibility"""
return self._registered_prompts
def discover_prompts(self) -> List[BaseServerPrompt]:
"""Discover all prompts in the prompts directory"""
prompts = []
prompts_dir = Path(self.prompts_directory)
if not prompts_dir.exists():
return prompts
# Find all Python files in the prompts directory
for file_path in prompts_dir.glob("*.py"):
if file_path.name.startswith("__"):
continue
module_name = file_path.stem
try:
# Import the module
spec = importlib.util.spec_from_file_location(module_name, file_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
# Find all BaseServerPrompt subclasses in the module
for name, obj in inspect.getmembers(module, inspect.isclass):
if (issubclass(obj, BaseServerPrompt) and
obj != BaseServerPrompt and
not inspect.isabstract(obj)):
try:
prompt_instance = obj()
prompts.append(prompt_instance)
self._registered_prompts[prompt_instance.name] = prompt_instance
except Exception as e:
print(f"Warning: Could not instantiate prompt {name}: {e}")
except Exception as e:
print(f"Warning: Could not load prompt file {file_path}: {e}")
return prompts
def register_prompt(self, prompt: BaseServerPrompt) -> None:
"""Register a single prompt"""
self._registered_prompts[prompt.name] = prompt
def register_prompts(self, prompts: List[BaseServerPrompt]) -> None:
"""Register multiple prompts"""
for prompt in prompts:
self.register_prompt(prompt)
def get_prompt(self, name: str) -> Optional[BaseServerPrompt]:
"""Get a prompt by name"""
return self._registered_prompts.get(name)
def get_all_prompts(self) -> List[BaseServerPrompt]:
"""Get all registered prompts"""
return list(self._registered_prompts.values())
def get_prompt_names(self) -> List[str]:
"""Get all registered prompt names"""
return list(self._registered_prompts.keys())
def register_prompts_with_server(self, mcp_server) -> None:
"""Register all discovered prompts with a FastMCP server"""
# First discover prompts if not already done
if not self._registered_prompts:
self.discover_prompts()
# Register each prompt with the server
for prompt in self._registered_prompts.values():
prompt.create_fastmcp_prompt(mcp_server)
def clear_prompts(self) -> None:
"""Clear all registered prompts"""
self._registered_prompts.clear()