Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/threat_profiles.py:22-51 (handler)

  Main handler function for list_threat_profiles tool. Decorated with @server.tool() to register as MCP tool. Takes Context, limit (default 10), and optional api_key parameters. Uses vt_client to call /threat_profiles endpoint and returns sanitized list of threat profile dictionaries.

  @server.tool()
  async def list_threat_profiles(
      ctx: Context, limit: int = 10, api_key: str = None
  ) -> typing.List[typing.Dict[str, typing.Any]]:
    """List your Threat Profiles at Google Threat Intelligence.
  
    Threat Profiles filter all of Google TI's threat intelligence
    so you can focus only on the threats that matter most
    to your organization.
  
    Threat Profiles let you apply top-level filters
    for Target Industries and Target Regions to immediately provide
    a more focused view of relevant threats.
  
    When searching for threats, we must use this tool first to check
    if there is any Threat Profile that matches the user query
    before peforming a general search using the `search_threats` tool.
  
    Recommendations from Threat Profiles are more relevants to users
    than generic search threats. Use them as long as
    they match user's query.
  
    Returns:
      List of Threat Profiles.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      res = await utils.consume_vt_iterator(
          client, "/threat_profiles", limit=limit
      )
    return utils.sanitize_response([o.to_dict() for o in res])

• gti_mcp/tools/threat_profiles.py:22-22 (registration)

  Tool registration via @server.tool() decorator. This decorator from FastMCP registers the function as an available MCP tool on the server instance defined in server.py.

  @server.tool()

• gti_mcp/utils.py:20-26 (helper)

  Helper function consume_vt_iterator used by list_threat_profiles to consume a vt.Iterator and return a list of objects from the VirusTotal API.

  async def consume_vt_iterator(
      vt_client: vt.Client, endpoint: str, params: dict | None = None, limit: int = 10):
    """Consumes a vt.Iterator iterator and return the list of objects."""
    res = []
    async for obj in vt_client.iterator(endpoint, params=params, limit=limit):
      res.append(obj)
    return res

• gti_mcp/utils.py:119-138 (helper)

  Helper function sanitize_response used by list_threat_profiles to recursively remove empty dictionaries and lists from API responses before returning to the client.

  def sanitize_response(data: typing.Any) -> typing.Any:
    """Removes empty dictionaries and lists recursively from a response."""
    if isinstance(data, dict):
      sanitized_dict = {}
      for key, value in data.items():
        sanitized_value = sanitize_response(value)
        if sanitized_value is not None:
          sanitized_dict[key] = sanitized_value
      return sanitized_dict
    elif isinstance(data, list):
      sanitized_list = []
      for item in data:
        sanitized_item = sanitize_response(item)
        if sanitized_item is not None:
          sanitized_list.append(sanitized_item)
      return sanitized_list
    elif isinstance(data, str):
      return data if data else None
    else:
      return data

• gti_mcp/tools/threat_profiles.py:23-25 (schema)

  Function signature defines input/output schema: takes Context, limit (int, default 10), api_key (str, optional) and returns List[Dict[str, Any]] representing threat profiles.

  async def list_threat_profiles(
      ctx: Context, limit: int = 10, api_key: str = None
  ) -> typing.List[typing.Dict[str, typing.Any]]: