"""
MCP PostgreSQL Server - Main entry point.
This is the main FastMCP server that exposes user resolution and calendar tools.
Business logic is in src/user_resolver.py and src/calendar_resolver.py.
Database operations are in src/database.py and src/database_calendar.py.
Run with:
- STDIO: fastmcp dev server.py
- HTTP: python server.py --http
"""
import logging
import os
from contextlib import asynccontextmanager
from typing import Any, AsyncIterator
from dotenv import load_dotenv
from fastmcp import FastMCP
from fastmcp.server.auth.providers.jwt import StaticTokenVerifier
# Import from our modules
from src.database import init_pool, close_pool
from src.user_resolver import (
resolve_user_impl,
resolve_users_batch_impl,
confirm_user_impl,
confirm_users_batch_impl,
# Team management tools
get_team_members_impl,
get_manager_of_user_impl,
get_user_by_uuid_impl,
resolve_user_in_team_impl,
)
from src.calendar_resolver import (
get_user_calendar_insights_impl,
query_user_meetings_impl,
get_meeting_details_impl,
)
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
# === AUTHENTICATION ===
_auth_token = os.getenv("MCP_AUTH_TOKEN")
_auth = None
if _auth_token:
_auth = StaticTokenVerifier(
tokens={
_auth_token: {
"client_id": "mcp-client",
"scopes": ["read", "write"]
}
}
)
logger.info("Authentication ENABLED - Bearer token required")
else:
logger.warning("Authentication DISABLED - No MCP_AUTH_TOKEN set in .env")
# === LIFESPAN ===
@asynccontextmanager
async def lifespan(mcp: FastMCP) -> AsyncIterator[None]:
"""Manage server lifecycle - initialize and cleanup database pool."""
logger.info("Starting MCP PostgreSQL Server...")
try:
await init_pool()
yield
except Exception as e:
logger.error(f"Failed to initialize database: {e}")
raise
finally:
await close_pool()
logger.info("Server shutdown complete")
# === CREATE SERVER ===
mcp = FastMCP(
name="mcp-postgres-server",
lifespan=lifespan,
auth=_auth,
)
# === MCP TOOLS ===
@mcp.tool()
async def resolve_user(name_or_email: str) -> dict[str, Any]:
"""
Resolve a user name or email to their UUID.
Given a name or email, attempts to find the matching user.
Returns the user's UUID if confidently resolved, or asks for
verification/disambiguation if uncertain.
Args:
name_or_email: The user's name or email to resolve. Can be full name,
partial name, email, or email prefix.
Returns:
Dictionary with:
- action: One of "resolved", "verify", "disambiguate", "not_found"
- For "resolved": user_id, email, name, confidence
- For "verify"/"disambiguate": candidates list with email/name for verification
- For "not_found": message suggesting alternatives
"""
return await resolve_user_impl(name_or_email)
@mcp.tool()
async def resolve_users_batch(names: list[str]) -> dict[str, Any]:
"""
Resolve multiple user names or emails to their UUIDs in a single call.
Efficiently resolves a batch of names/emails and returns results for each.
Args:
names: List of names or emails to resolve. Maximum 50 items.
Returns:
Dictionary with:
- summary: Counts of resolved, verify_needed, disambiguate_needed, not_found
- resolved_uuids: Dict mapping input names to UUIDs (only for resolved)
- results: List of individual results for each input
- all_resolved: Boolean indicating if all inputs were resolved
"""
return await resolve_users_batch_impl(names)
@mcp.tool()
async def confirm_user(email: str) -> dict[str, Any]:
"""
Confirm a user by their exact email and get their UUID.
Use this after verification/disambiguation to get the UUID for a confirmed email.
Args:
email: The exact email address to look up.
Returns:
Dictionary with:
- action: "resolved" or "not_found"
- For "resolved": user_id, email, name
- For "not_found": message
"""
return await confirm_user_impl(email)
@mcp.tool()
async def confirm_users_batch(emails: list[str]) -> dict[str, Any]:
"""
Confirm multiple users by their exact emails and get their UUIDs.
Use this after verification/disambiguation when you have multiple
emails to confirm. More efficient than calling confirm_user multiple times.
Args:
emails: List of exact email addresses to look up. Maximum 50 items.
Returns:
Dictionary with:
- summary: Counts of resolved and not_found
- resolved_uuids: Dict mapping emails to UUIDs
- results: List of individual results for each email
- all_resolved: Boolean indicating if all emails were resolved
"""
return await confirm_users_batch_impl(emails)
# === TEAM MANAGEMENT TOOLS ===
@mcp.tool()
async def get_team_members(manager_identifier: str) -> dict[str, Any]:
"""
Get all team members under a specific manager.
Retrieves all users who report to the specified manager (including
users under them as remote manager).
Args:
manager_identifier: Manager's email (preferred), UUID, or name.
Email is most reliable for exact matches.
Returns:
Dictionary with:
- action: "resolved" or "not_found" or "error"
- team_count: Number of team members
- members: List of team members with uuid, name, email
Example:
get_team_members("john.manager@company.com")
get_team_members("550e8400-e29b-41d4-a716-446655440000")
"""
return await get_team_members_impl(manager_identifier)
@mcp.tool()
async def get_manager_of_user(user_identifier: str) -> dict[str, Any]:
"""
Get the manager(s) of a specific user.
Returns both primary manager and remote manager if applicable.
Args:
user_identifier: User's email (preferred) or UUID.
Returns:
Dictionary with:
- action: "resolved" or "not_found" or "error"
- user: User details (uuid, name, email)
- managers: List of primary managers (uuid, name, email)
- remote_managers: List of remote managers (uuid, name, email)
- has_manager: Boolean indicating if user has a primary manager
- has_remote_manager: Boolean indicating if user has a remote manager
Example:
get_manager_of_user("employee@company.com")
"""
return await get_manager_of_user_impl(user_identifier)
@mcp.tool()
async def get_user_by_uuid(user_uuid: str) -> dict[str, Any]:
"""
Get complete user details by their UUID.
Use this for quick lookups when you already have the UUID from
a previous resolution.
Args:
user_uuid: The user's UUID (format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx)
Returns:
Dictionary with:
- action: "resolved" or "not_found" or "error"
- user: User details (uuid, name, email)
- managers: List of primary managers
- remote_managers: List of remote managers
- clients: List of clients the user belongs to
Example:
get_user_by_uuid("550e8400-e29b-41d4-a716-446655440000")
"""
return await get_user_by_uuid_impl(user_uuid)
@mcp.tool()
async def resolve_user_in_team(
name_or_email: str,
manager_identifier: str
) -> dict[str, Any]:
"""
Resolve a user name/email within a specific manager's team only.
This is a SCOPED search that only returns users who report to the
specified manager. Use this when you want to ensure searches stay
within team boundaries (security/relevance).
Args:
name_or_email: The user's name or email to search for.
manager_identifier: Manager's email (preferred) or UUID to scope the search.
Returns:
Dictionary with:
- action: "resolved", "verify", "disambiguate", "not_found", or "error"
- For "resolved": user_id, email, name, confidence, scope
- For "verify"/"disambiguate": candidates list, scope
- scope: Indicates which team was searched
Example:
resolve_user_in_team("john", "team.manager@company.com")
resolve_user_in_team("john.doe@company.com", "550e8400-e29b-41d4-a716-446655440000")
"""
return await resolve_user_in_team_impl(name_or_email, manager_identifier)
# === CALENDAR INSIGHTS TOOLS ===
@mcp.tool()
async def get_user_calendar_insights(
user_identifier: str,
date: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
include_daily: bool = False,
include_meetings: bool = False,
) -> dict[str, Any]:
"""
Get comprehensive calendar insights for a user - the complete dashboard.
Provides health assessment, metrics, statistical extremes (longest/shortest/
largest meetings), recurring meeting analysis, and quality metrics.
Args:
user_identifier: User's email (preferred) or UUID.
date: Single date (YYYY-MM-DD) for day view. If provided, ignores start/end.
start_date: Range start (YYYY-MM-DD). Defaults to last 7 days if not set.
end_date: Range end (YYYY-MM-DD). Max range: 90 days.
include_daily: Include daily breakdown array (default: false).
include_meetings: Include list of actual meetings (default: false).
Returns:
Dictionary with:
- action: "resolved", "no_data", "not_found", or "error"
- user: User details (uuid, name, email)
- period: Date range and working days count
- health: Status (healthy/warning/at_risk), concerns, positives, suggestions
- time: Total meeting hours, focus hours, percentages
- averages: Per-day metrics (meeting load %, focus minutes, meetings/day)
- counts: Total meetings, high load days, low focus days
- by_type: Breakdown by meeting type (1:1, standup, review, etc.)
- recurring: Recurring meeting count, percentage, top series
- external: External meeting count and hours
- quality: Agenda coverage, average quality, large meetings count
- extremes: Longest/shortest/largest meetings, busiest/lightest days
- daily: (optional) Array of daily metrics
- meetings: (optional) Array of meeting details
Example:
get_user_calendar_insights("john@company.com", date="2025-12-12")
get_user_calendar_insights("john@company.com", start_date="2025-12-01", end_date="2025-12-31")
get_user_calendar_insights("john@company.com", include_meetings=True)
"""
return await get_user_calendar_insights_impl(
user_identifier, date, start_date, end_date, include_daily, include_meetings
)
@mcp.tool()
async def query_user_meetings(
user_identifier: str,
start_date: str,
end_date: str,
sort_by: str = "start_time",
order: str = "desc",
limit: int = 20,
meeting_type: str | None = None,
is_external: bool | None = None,
is_recurring: bool | None = None,
has_agenda: bool | None = None,
min_duration: int | None = None,
min_attendees: int | None = None,
search: str | None = None,
) -> dict[str, Any]:
"""
Query user's meetings with flexible filtering and sorting.
Use this to find specific meetings, get sorted lists (longest, shortest,
largest), filter by criteria, or search by title keyword.
Args:
user_identifier: User's email (preferred) or UUID.
start_date: Start date (YYYY-MM-DD).
end_date: End date (YYYY-MM-DD).
sort_by: Sort field - "start_time", "duration", "attendees", "agenda_quality".
order: Sort order - "asc" or "desc" (default: desc).
limit: Max results (default: 20, max: 100).
meeting_type: Filter by type (1_1, STANDUP, REVIEW, PLANNING, EXTERNAL, OTHER).
is_external: Filter by external flag (true/false).
is_recurring: Filter by recurring flag (true/false).
has_agenda: Filter by agenda presence (true/false).
min_duration: Minimum duration in minutes.
min_attendees: Minimum attendee count.
search: Title keyword search (case-insensitive).
Returns:
Dictionary with:
- action: "resolved", "not_found", or "error"
- user: User details
- query: Period, sort, filters applied
- total_found: Number of meetings matching criteria
- meetings: Array of meeting details (title, date, duration, type, etc.)
Example:
query_user_meetings("john@company.com", "2025-12-01", "2025-12-31", sort_by="duration", limit=1)
query_user_meetings("john@company.com", "2025-12-01", "2025-12-31", search="sprint planning")
query_user_meetings("john@company.com", "2025-12-01", "2025-12-31", is_recurring=True)
query_user_meetings("john@company.com", "2025-12-01", "2025-12-31", has_agenda=False)
"""
return await query_user_meetings_impl(
user_identifier, start_date, end_date, sort_by, order, limit,
meeting_type, is_external, is_recurring, has_agenda,
min_duration, min_attendees, search
)
@mcp.tool()
async def get_meeting_details(event_id: str) -> dict[str, Any]:
"""
Get full details of a specific meeting.
Use the event_id from query_user_meetings or get_user_calendar_insights
to get complete information including attendee list, organizer, and
agenda quality signals.
Args:
event_id: The event ID from a previous query.
Returns:
Dictionary with:
- action: "resolved", "not_found", or "error"
- event_id: The event ID
- title: Meeting title
- organizer: Organizer's email
- time: Start, end, date, duration_min
- attendees: Total count, internal/external counts, full list
- classification: Meeting type, is_recurring, is_external, is_large_meeting
- quality: has_agenda, agenda_quality_index, agenda_signals
- recurring_info: series_id and instance_key (if recurring)
- tagged_priorities: Priority tags from title/description
Example:
get_meeting_details("abc123_20251212T100000Z")
"""
return await get_meeting_details_impl(event_id)
# === HTTP APP WITH CORS (for MCP Inspector) ===
def create_http_app():
"""Create HTTP app with CORS middleware for browser-based clients like MCP Inspector."""
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
middleware = [
Middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["GET", "POST", "DELETE", "OPTIONS"],
allow_headers=[
"mcp-protocol-version",
"mcp-session-id",
"Authorization",
"Content-Type",
],
expose_headers=["mcp-session-id"],
)
]
return mcp.http_app(middleware=middleware)
# === ENTRY POINT ===
if __name__ == "__main__":
import sys
if "--http" in sys.argv:
# HTTP mode with CORS
port = int(os.getenv("MCP_PORT", "8000"))
host = os.getenv("MCP_HOST", "0.0.0.0")
logger.info(f"Starting HTTP server on {host}:{port}")
logger.info(f"MCP endpoint: http://{host}:{port}/mcp")
logger.info("CORS enabled for MCP Inspector")
import uvicorn
app = create_http_app()
uvicorn.run(app, host=host, port=port)
else:
# Default STDIO mode
mcp.run()
