Finizi B4B MCP Server

"""Entity management tools for B4B MCP server.""" from typing import Optional import httpx import structlog from mcp.server.fastmcp import Context from ..auth.token_handler import extract_user_token from ..client.api_client import get_api_client from ..server import mcp from ..utils.validators import validate_uuid, validate_page_params logger = structlog.get_logger(__name__) api_client = get_api_client() @mcp.tool() async def list_entities( page: int = 1, per_page: int = 20, search: Optional[str] = None, ctx: Context = None ) -> dict: """ List entities the user has access to. Args: page: Page number (default: 1) per_page: Items per page, max 100 (default: 20) search: Optional search query for entity name or tax ID ctx: MCP context (automatically provided) Returns: { "items": [...], "total": int, "page": int, "per_page": int, "pages": int } """ log = logger.bind(action="list_entities", page=page, per_page=per_page) log.info("Listing entities") try: # Extract user token user_token = await extract_user_token(ctx) # Validate pagination parameters page, per_page = validate_page_params(page, per_page) # Build query parameters params = { "page": page, "per_page": per_page } if search: params["search"] = search # Call B4B API response_data = await api_client.get( "/entities/", token=user_token, params=params ) log.info( "Entities listed successfully", total=response_data.get("total", 0), page=response_data.get("page", page) ) return response_data except ValueError as e: # From extract_user_token or validators log.error(f"Validation error: {str(e)}") return {"error": str(e)} except httpx.HTTPStatusError as e: if e.response.status_code == 401: log.error("Authentication failed") return {"error": "Authentication failed. Please login first."} elif e.response.status_code == 403: log.error("Access forbidden") return {"error": "You don't have permission to list entities"} else: try: detail = e.response.json().get("detail", str(e)) except Exception: detail = str(e) log.error(f"API error: {detail}", status_code=e.response.status_code) return {"error": f"API error: {detail}"} except Exception as e: log.error(f"Unexpected error: {str(e)}") return {"error": str(e)} @mcp.tool() async def get_entity(entity_id: str, ctx: Context) -> dict: """ Get detailed information about a specific entity. Args: entity_id: UUID of the entity ctx: MCP context (automatically provided) Returns: Complete entity details including tax information, address, etc. """ log = logger.bind(action="get_entity", entity_id=entity_id) log.info("Getting entity details") try: # Extract user token user_token = await extract_user_token(ctx) # Validate entity ID entity_id = validate_uuid(entity_id, "entity_id") # Call B4B API response_data = await api_client.get( f"/entities/{entity_id}", token=user_token ) log.info("Entity retrieved successfully", entity_name=response_data.get("name")) return response_data except ValueError as e: # From extract_user_token or validators log.error(f"Validation error: {str(e)}") return {"error": str(e)} except httpx.HTTPStatusError as e: if e.response.status_code == 403: log.error("Access forbidden", entity_id=entity_id) return {"error": f"You don't have access to entity {entity_id}"} elif e.response.status_code == 404: log.error("Entity not found", entity_id=entity_id) return {"error": f"Entity {entity_id} not found"} else: try: detail = e.response.json().get("detail", str(e)) except Exception: detail = str(e) log.error(f"API error: {detail}", status_code=e.response.status_code) return {"error": f"API error: {detail}"} except Exception as e: log.error(f"Unexpected error: {str(e)}") return {"error": str(e)} @mcp.tool() async def create_entity( name: str, tax_id: str, entity_type: str, address: Optional[str] = None, phone: Optional[str] = None, email: Optional[str] = None, ctx: Context = None ) -> dict: """ Create a new entity. Args: name: Entity name (company name or individual name) tax_id: Tax identification number (10 or 13 digits) entity_type: "company" or "individual" address: Optional business address phone: Optional phone number email: Optional email address ctx: MCP context (automatically provided) Returns: Created entity details including generated ID """ log = logger.bind(action="create_entity", name=name, entity_type=entity_type) log.info("Creating entity") try: # Extract user token user_token = await extract_user_token(ctx) # Build payload with required fields payload = { "name": name, "tax_id": tax_id, "entity_type": entity_type } # Add optional fields if provided if address is not None: payload["address"] = address if phone is not None: payload["phone"] = phone if email is not None: payload["email"] = email # Call B4B API response_data = await api_client.post( "/entities/", token=user_token, json=payload ) log.info( "Entity created successfully", entity_id=response_data.get("id"), entity_name=response_data.get("name") ) return response_data except ValueError as e: # From extract_user_token or validators log.error(f"Validation error: {str(e)}") return {"error": str(e)} except httpx.HTTPStatusError as e: if e.response.status_code == 400: try: detail = e.response.json().get("detail", str(e)) except Exception: detail = str(e) log.error(f"Validation error from API: {detail}") return {"error": f"Validation error: {detail}"} elif e.response.status_code == 403: log.error("Access forbidden") return {"error": "You don't have permission to create entities"} else: try: detail = e.response.json().get("detail", str(e)) except Exception: detail = str(e) log.error(f"API error: {detail}", status_code=e.response.status_code) return {"error": f"API error: {detail}"} except Exception as e: log.error(f"Unexpected error: {str(e)}") return {"error": str(e)} @mcp.tool() async def update_entity( entity_id: str, name: Optional[str] = None, address: Optional[str] = None, phone: Optional[str] = None, email: Optional[str] = None, ctx: Context = None ) -> dict: """ Update entity information. Args: entity_id: UUID of entity to update name: Optional new name address: Optional new address phone: Optional new phone email: Optional new email ctx: MCP context (automatically provided) Returns: Updated entity details """ log = logger.bind(action="update_entity", entity_id=entity_id) log.info("Updating entity") try: # Extract user token user_token = await extract_user_token(ctx) # Validate entity ID entity_id = validate_uuid(entity_id, "entity_id") # Build payload with ONLY provided fields (skip None values) payload = {} if name is not None: payload["name"] = name if address is not None: payload["address"] = address if phone is not None: payload["phone"] = phone if email is not None: payload["email"] = email # Check if there are any fields to update if not payload: log.warning("No fields to update") return {"error": "No fields provided for update"} # Call B4B API response_data = await api_client.put( f"/entities/{entity_id}", token=user_token, json=payload ) log.info( "Entity updated successfully", entity_id=entity_id, updated_fields=list(payload.keys()) ) return response_data except ValueError as e: # From extract_user_token or validators log.error(f"Validation error: {str(e)}") return {"error": str(e)} except httpx.HTTPStatusError as e: if e.response.status_code == 403: log.error("Access forbidden", entity_id=entity_id) return {"error": f"You don't have access to entity {entity_id}"} elif e.response.status_code == 404: log.error("Entity not found", entity_id=entity_id) return {"error": f"Entity {entity_id} not found"} elif e.response.status_code == 400: try: detail = e.response.json().get("detail", str(e)) except Exception: detail = str(e) log.error(f"Validation error from API: {detail}") return {"error": f"Validation error: {detail}"} else: try: detail = e.response.json().get("detail", str(e)) except Exception: detail = str(e) log.error(f"API error: {detail}", status_code=e.response.status_code) return {"error": f"API error: {detail}"} except Exception as e: log.error(f"Unexpected error: {str(e)}") return {"error": str(e)}