"""Company rating tool for Birre."""
import logging
from typing import Dict, Any, Optional
from fastmcp import FastMCP, Context
import bitsight
try:
from ..config import get_config
except ImportError:
from birre.config import get_config
logger = logging.getLogger(__name__)
async def get_company_rating_impl(ctx: Context, guid: str) -> Dict[str, Any]:
"""
Get security rating for a company by GUID.
Automatically manages BitSight subscription if needed:
- If not subscribed: subscribes with continuous_monitoring, gets rating, then unsubscribes
- If already subscribed: gets rating directly
- Returns rating data with operation flags
Args:
guid: BitSight GUID of the company (required)
Returns:
Dictionary containing:
- rating: Current security rating (0-900 scale)
- rating_date: Date of the rating
- grade: Letter grade representation
- auto_subscribed: Boolean indicating if auto-subscription was performed
- auto_unsubscribed: Boolean indicating if auto-unsubscription was performed
- company_name: Name of the company (if available)
- error: Error message if operation failed
"""
auto_subscribed = False
auto_unsubscribed = False
company_name = None
try:
# Validate GUID
if not guid or not isinstance(guid, str):
return {
"error": "Valid company GUID is required",
"rating": None,
"rating_date": None,
"grade": None,
"auto_subscribed": False,
"auto_unsubscribed": False,
}
await ctx.info(f"Getting rating for company GUID: {guid}")
# Initialize BitSight API clients
companies_api = bitsight.Companies()
subscriptions_api = bitsight.Subscriptions()
# First, try to get company details to check current subscription status
try:
await ctx.debug("Checking current subscription status...")
company_details = companies_api.get_company_details(guid)
company_name = company_details.get("name")
# Check for current_rating and get latest rating date from ratings array
current_rating = company_details.get("current_rating")
ratings_history = company_details.get("ratings", [])
# Get the most recent rating date from ratings history
rating_date = None
if ratings_history:
rating_date = ratings_history[0].get(
"rating_date"
) # First entry is most recent
if current_rating is not None:
await ctx.info(f"Company already subscribed, rating: {current_rating}")
# Calculate letter grade from numeric rating
grade = _calculate_letter_grade(current_rating)
return {
"rating": current_rating,
"rating_date": rating_date,
"grade": grade,
"company_name": company_name,
"auto_subscribed": False,
"auto_unsubscribed": False,
}
else:
await ctx.info(
"No current rating data found, subscription may be needed"
)
except Exception as e:
await ctx.debug(
f"Could not get company details (may need subscription): {str(e)}"
)
# If we reach here, we likely need to subscribe
await ctx.info(
"Attempting to subscribe to company for continuous monitoring..."
)
try:
# Subscribe with continuous monitoring (default)
subscribe_result = subscriptions_api.post_subscribe(guid)
auto_subscribed = True
await ctx.info("Successfully subscribed to company")
# Now try to get the rating
company_details = companies_api.get_company_details(guid)
company_name = company_details.get("name", company_name)
# Use current_rating
current_rating = company_details.get("current_rating")
ratings_history = company_details.get("ratings", [])
# Get the most recent rating date from ratings history
rating_date = None
if ratings_history:
rating_date = ratings_history[0].get(
"rating_date"
) # First entry is most recent
if current_rating is None:
await ctx.warning(
"Rating data not immediately available after subscription"
)
return {
"error": "Rating data not available immediately after subscription. Please try again shortly.",
"rating": None,
"rating_date": None,
"grade": None,
"company_name": company_name,
"auto_subscribed": auto_subscribed,
"auto_unsubscribed": False,
}
# Calculate letter grade
grade = _calculate_letter_grade(current_rating)
# Now unsubscribe as per requirements for ephemeral subscription
try:
await ctx.info(
"Unsubscribing from company as per ephemeral subscription model..."
)
unsubscribe_result = subscriptions_api.delete_unsubscribe(guid)
auto_unsubscribed = True
await ctx.info("Successfully unsubscribed from company")
except Exception as unsub_error:
await ctx.warning(
f"Could not unsubscribe automatically: {str(unsub_error)}"
)
await ctx.warning("Manual cleanup may be required")
return {
"rating": current_rating,
"rating_date": rating_date,
"grade": grade,
"company_name": company_name,
"auto_subscribed": auto_subscribed,
"auto_unsubscribed": auto_unsubscribed,
}
except Exception as sub_error:
error_msg = f"Failed to subscribe to company: {str(sub_error)}"
await ctx.error(error_msg)
return {
"error": error_msg,
"rating": None,
"rating_date": None,
"grade": None,
"company_name": company_name,
"auto_subscribed": False,
"auto_unsubscribed": False,
}
except Exception as e:
error_msg = f"Failed to get company rating: {str(e)}"
await ctx.error(error_msg)
logger.error(error_msg, exc_info=True)
return {
"error": error_msg,
"rating": None,
"rating_date": None,
"grade": None,
"company_name": company_name,
"auto_subscribed": auto_subscribed,
"auto_unsubscribed": auto_unsubscribed,
}
def _calculate_letter_grade(rating: int) -> str:
"""
Convert numeric BitSight rating to letter grade.
Args:
rating: Numeric rating (0-900)
Returns:
Letter grade (A, B, C, D, F)
"""
if rating >= 740:
return "A"
elif rating >= 640:
return "B"
elif rating >= 560:
return "C"
elif rating >= 480:
return "D"
else:
return "F"
def register_company_rating_tool(server: FastMCP) -> None:
"""Register the company rating tool with the FastMCP server."""
@server.tool
async def get_company_rating(ctx: Context, guid: str) -> Dict[str, Any]:
"""Get security rating for a company by GUID."""
return await get_company_rating_impl(ctx, guid)
