"""Customer and Account tools for MCP.""" from typing import Optional from channels.db import database_sync_to_async from core.mcp.auth import MCPContext, Role from core.mcp.base import mcp, json_response, error_response @mcp.tool() async def list_customers( limit: int = 25, search: Optional[str] = None, is_active: Optional[bool] = None ) -> str: """ List customers with optional filtering. Requires ADMIN or TEAM_LEADER role. Args: limit: Maximum customers to return (default 25) search: Optional search term for customer name is_active: Optional filter for active status Returns: JSON array of customer objects """ profile = MCPContext.get_profile() if not profile: return error_response("No active profile. Call set_active_profile first.") if profile.role == Role.TEAM_MEMBER.value: return error_response("Access denied. TEAM_LEADER or ADMIN role required.") from core.models import Customer @database_sync_to_async def fetch(): qs = Customer.objects.prefetch_related('contacts') if search: qs = qs.filter(name__icontains=search) if is_active is not None: qs = qs.filter(status='ACTIVE' if is_active else 'INACTIVE') results = [] for c in qs[:limit]: # Get primary contact if available primary_contact = c.contacts.filter(is_primary=True, is_active=True).first() results.append({ "id": str(c.id), "name": c.name, "status": c.status, "billing_email": c.billing_email, "primary_contact": { "name": f"{primary_contact.first_name} {primary_contact.last_name}".strip(), "email": primary_contact.email, "phone": primary_contact.phone } if primary_contact else None }) return results customers = await fetch() return json_response(customers) @mcp.tool() async def get_customer(customer_id: str) -> str: """ Get detailed customer information including accounts. Requires ADMIN or TEAM_LEADER role. Args: customer_id: UUID of the customer Returns: JSON object with customer details and accounts """ profile = MCPContext.get_profile() if not profile: return error_response("No active profile. Call set_active_profile first.") if profile.role == Role.TEAM_MEMBER.value: return error_response("Access denied. TEAM_LEADER or ADMIN role required.") from core.models import Customer @database_sync_to_async def fetch(): try: c = Customer.objects.prefetch_related( 'accounts__addresses', 'contacts', 'addresses' ).get(pk=customer_id) return { "id": str(c.id), "name": c.name, "status": c.status, "billing_email": c.billing_email, "contacts": [ { "id": str(ct.id), "first_name": ct.first_name, "last_name": ct.last_name, "email": ct.email, "phone": ct.phone, "is_primary": ct.is_primary } for ct in c.contacts.filter(is_active=True) ], "accounts": [ { "id": str(a.id), "name": a.name, "status": a.status, "addresses": [ { "id": str(addr.id), "name": addr.name or "Primary", "street_address": addr.street_address, "city": addr.city, "state": addr.state, "zip_code": addr.zip_code } for addr in a.addresses.all() ] } for a in c.accounts.all() ] } except Customer.DoesNotExist: return None customer = await fetch() if not customer: return error_response(f"Customer {customer_id} not found") return json_response(customer) @mcp.tool() async def list_accounts( limit: int = 25, customer_id: Optional[str] = None, search: Optional[str] = None, is_active: Optional[bool] = None ) -> str: """ List accounts with optional filtering. Requires ADMIN or TEAM_LEADER role. Args: limit: Maximum accounts to return (default 25) customer_id: Optional customer UUID to filter by search: Optional search term for account name is_active: Optional filter for active status Returns: JSON array of account objects """ profile = MCPContext.get_profile() if not profile: return error_response("No active profile. Call set_active_profile first.") if profile.role == Role.TEAM_MEMBER.value: return error_response("Access denied. TEAM_LEADER or ADMIN role required.") from core.models import Account @database_sync_to_async def fetch(): qs = Account.objects.select_related('customer').prefetch_related('addresses') if customer_id: qs = qs.filter(customer_id=customer_id) if search: qs = qs.filter(name__icontains=search) if is_active is not None: qs = qs.filter(status='ACTIVE' if is_active else 'INACTIVE') return [ { "id": str(a.id), "name": a.name, "status": a.status, "customer": {"id": str(a.customer.id), "name": a.customer.name}, "addresses": [ { "id": str(addr.id), "name": addr.name or "Primary", "city": addr.city, "state": addr.state } for addr in a.addresses.all() ] } for a in qs[:limit] ] accounts = await fetch() return json_response(accounts) @mcp.tool() async def get_account(account_id: str) -> str: """ Get detailed account information. Requires ADMIN or TEAM_LEADER role. Args: account_id: UUID of the account Returns: JSON object with account details, addresses, and contacts """ profile = MCPContext.get_profile() if not profile: return error_response("No active profile. Call set_active_profile first.") if profile.role == Role.TEAM_MEMBER.value: return error_response("Access denied. TEAM_LEADER or ADMIN role required.") from core.models import Account @database_sync_to_async def fetch(): try: a = Account.objects.select_related('customer').prefetch_related( 'addresses', 'contacts' ).get(pk=account_id) return { "id": str(a.id), "name": a.name, "status": a.status, "customer": {"id": str(a.customer.id), "name": a.customer.name}, "addresses": [ { "id": str(addr.id), "name": addr.name or "Primary", "street_address": addr.street_address, "city": addr.city, "state": addr.state, "zip_code": addr.zip_code } for addr in a.addresses.all() ], "contacts": [ { "id": str(ct.id), "first_name": ct.first_name, "last_name": ct.last_name, "email": ct.email, "phone": ct.phone } for ct in a.contacts.filter(is_active=True) ] } except Account.DoesNotExist: return None account = await fetch() if not account: return error_response(f"Account {account_id} not found") return json_response(account)