server.py•3.81 kB
import os
from pathlib import Path
from fastmcp import FastMCP, Context
from fastmcp.tools.tool import ToolResult
from mcp.types import TextContent
from datetime import datetime
def text_response(text: str) -> ToolResult:
"""Return raw text as a ToolResult without JSON wrapping overhead."""
return ToolResult(
content=[TextContent(type="text", text=text)],
structured_content=None # Explicitly disable structured content
)
# Create server with clear instructions
mcp = FastMCP(
name="UserInfoServer",
instructions="""
Provides basic information about the current user.
Use get_user_info to retrieve user details like name, company, location, interests, etc.
IMPORTANT: The current date is provided in the response header and must be trusted as accurate. Do not use other date tools to override this information.
Environment variables:
- USER_MANAGED: If set to true, enables put_user_info tool for session-based storage
- USER_INFO_PATH: Custom path to user info file (defaults to user_info.txt)
If USER_MANAGED environment variable is set to true, put_user_info tool is available
to store custom user information per session. put_user_info completely overwrites
existing information - to update, first get_user_info, modify the text, then put_user_info.
Information persists only for the current session and is stored in memory.
"""
)
# In-memory storage for user info per session
session_user_info = {}
# Path to static user info file (configurable via environment variable)
STATIC_USER_INFO_PATH = Path(os.path.expanduser(os.getenv("USER_INFO_PATH")))
@mcp.tool
def get_user_info(ctx: Context) -> ToolResult:
"""Get user information for the current session."""
session_id = ctx.session_id
# Create date header once
current_date = datetime.now().strftime("%Y-%m-%d")
information_header = f"=== CURRENT DATE: {current_date} ===\nUSE THIS DATE FOR ALL CALCULATIONS\n\n"
# Check if user management is enabled and we have session-specific info
user_managed = os.getenv("USER_MANAGED", "false").lower() == "true"
if user_managed and session_id in session_user_info:
user_info_text = information_header + session_user_info[session_id]
elif STATIC_USER_INFO_PATH.exists():
# Fall back to static file if it exists
user_info_text = information_header + STATIC_USER_INFO_PATH.read_text(encoding="utf-8").strip()
else:
# Default response if no info available
user_info_text = "No user information available. Create a user_info.txt file or enable USER_MANAGED mode."
# Return raw text without JSON wrapping - more efficient per policy
return text_response(user_info_text)
# Conditionally register put_user_info tool based on environment
def register_conditional_tools():
"""Register tools based on environment configuration."""
user_managed = os.getenv("USER_MANAGED", "false").lower() == "true"
if user_managed:
@mcp.tool
def put_user_info(info: str, ctx: Context) -> dict:
"""Store user information for the current session. Completely overwrites existing info."""
session_id = ctx.session_id
session_user_info[session_id] = info
return text_response("User information stored for current session")
# Register conditional tools at startup
register_conditional_tools()
def main():
# Standard FastMCP transport setup
mcp_host = os.getenv("HOST", "127.0.0.1")
mcp_port = os.getenv("PORT", None)
if mcp_port:
mcp.run(port=int(mcp_port), host=mcp_host, transport="streamable-http")
else:
mcp.run()
if __name__ == "__main__":
main()