Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/files.py:228-248 (handler)

  Main handler function for get_file_behavior_summary tool. Retrieves file behavior summary from VirusTotal API, handles API errors and unexpected response formats, and returns sanitized data.

  @server.tool()
  async def get_file_behavior_summary(hash: str, ctx: Context, api_key: str = None) -> typing.Dict[str, typing.Any]:
    """Retrieve a summary of all the file behavior reports from all the sandboxes.
  
    Args:
      hash (required): MD5/SHA1/SHA256) hash that identifies the file.
    Returns:
      The file behavior summary.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      res = await client.get_async(f"/files/{hash}/behaviour_summary")
      res = await res.json_async()
  
    if "data" not in res:
        if "error" in res:
            logging.warning(f"VirusTotal API Error: {res['error']}")
            return {"error": f"VirusTotal API Error: {res['error']}"}
        logging.warning(f"Unexpected response format from VirusTotal API: {res}")
        return {"error": f"Unexpected response format from VirusTotal API: {res}"}
  
    return utils.sanitize_response(res["data"])

• gti_mcp/tools/files.py:228-228 (registration)

  Tool registration using @server.tool() decorator that registers get_file_behavior_summary as an MCP tool with the FastMCP server.

  @server.tool()

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

  Helper function sanitize_response that recursively removes empty dictionaries and lists from the response data 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/server.py:56-64 (helper)

  Async context manager vt_client that provides a vt.Client instance for API calls and ensures proper cleanup by closing the client after use.

  @asynccontextmanager
  async def vt_client(ctx: Context, api_key: str = None) -> AsyncIterator[vt.Client]:
    """Provides a vt.Client instance for the current request."""
    client = vt_client_factory(ctx, api_key)
  
    try:
      yield client
    finally:
      await client.close_async()